From 80a109235131a34b0395327d233f37502ac3890d Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:13:24 +0000 Subject: [PATCH] refactor(plan-selection): row (K) terms per path of edges, not a weight per candidate (#1203) Stage 3's accuracy weight moves from the approximate candidate to each path of edges (u, c, i) through which a query reaches it: one producer read by a ratio's numerator and denominator has a term for each, each checked by the composition alone, pairs of terms checked together. Row 8 and the warm start constrain and count the paths a plan depends on (d[q, t]). The per-candidate map (accuracy_weights, accuracy_positions.rs) is deleted; AccuracyWeight is now per term. Interim until #1207 step 6 replaces the weights with row (K)'s constant lambda_{q,c,i}. Corpus and sketch-plan snapshots unchanged. --- crates/plan-selection/src/groups.rs | 122 ++- .../src/groups/accuracy_positions.rs | 777 ------------------ .../src/groups/accuracy_terms.rs | 639 ++++++++++++++ crates/plan-selection/src/groups/tests.rs | 136 +-- crates/plan-selection/src/lib.rs | 2 +- crates/plan-selection/src/milp/model.rs | 216 +++-- crates/plan-selection/src/milp/model/tests.rs | 115 +-- crates/plan-selection/src/milp/warm_start.rs | 84 +- crates/plan-selection/src/report.rs | 2 +- 9 files changed, 1058 insertions(+), 1035 deletions(-) delete mode 100644 crates/plan-selection/src/groups/accuracy_positions.rs create mode 100644 crates/plan-selection/src/groups/accuracy_terms.rs diff --git a/crates/plan-selection/src/groups.rs b/crates/plan-selection/src/groups.rs index 1421f5d2a..8576ef4bd 100644 --- a/crates/plan-selection/src/groups.rs +++ b/crates/plan-selection/src/groups.rs @@ -11,9 +11,10 @@ //! 3. **Owners** ([`OwnerRegion`]): one per query and one per possible //! instance of each shareable candidate, each with its region of stateless //! candidates, the links inside it and the shareable candidates it may read. -//! 4. **Accuracy weights** ([`AccuracyWeight`]) per query and approximate -//! candidate, and pairs excluded together, from the composition of the -//! query's outputs. +//! 4. **Row (K) terms** ([`AccuracyTerm`]) per query: one per path of +//! edges `(u, c, i)` from an approximate producer `u` to the query's +//! output, each with its accuracy weight, and pairs of terms +//! excluded together, from the composition of the query's outputs. //! //! Nothing calls it yet; the MILP (C4) will. //! @@ -72,37 +73,46 @@ //! dominate each other the one with the smallest id is kept. Every link to //! a deleted candidate is deleted; its readers still read the dominating candidate. //! -//! ## Accuracy weights +//! ## Row (K) terms //! -//! `w(q, k)` says whether query `q` may read approximate candidate `k` (a -//! sketch [`ASAPOp::SummaryAgg`] or [`ASAPOp::SummaryTumblingWindowChain`], what -//! [`asap_accuracy::propagation::approximate_inputs`] counts), and pairs of -//! approximate candidates `q` may not read together. Both come from the -//! composition itself, never from an operator table -//! ([`accuracy_positions`]): `q`'s outputs are built over the candidates -//! with `k` approximate and every other candidate exact, their guarantees +//! Row (K) (#1204) bounds each query's composed error by a sum of terms, +//! one per approximate producer reaching the query. Until #1207 step 6 +//! gives each term a constant propagation factor, a term carries an +//! accuracy weight: one share of the query's budget, or unusable. Here a query `q` depends on an approximate +//! producer `u` (a sketch [`ASAPOp::SummaryAgg`] or +//! [`ASAPOp::SummaryTumblingWindowChain`], what +//! [`asap_accuracy::propagation::approximate_inputs`] counts) through a +//! path of edges `(u, c, i)` up to its output, and one producer may reach +//! `q` through several paths, e.g. a state read by both operands of a +//! ratio. So each term ([`AccuracyTerm`]) is one such path, with its own +//! accuracy weight ([`AccuracyWeight`]): `q` depends on the term when every +//! candidate on its path is selected where the path reads it. The weight comes +//! from the composition itself, never from an operator table +//! ([`accuracy_terms`]): `q`'s outputs are built with `u`'s values +//! approximate along the path and every other input exact, their guarantees //! composed by `asap-accuracy` ([`asap_accuracy::propagation::composed_guarantee`]) //! and checked by the same check the selected plan passes //! ([`crate::checks::guarantee_check`], with the deployment's model): -//! - every output exact: `k`'s values do not reach the output (only a row -//! count does), and `k` gets no weight; -//! - every output meets the target: `w(q, k) = 1`, one share of `q`'s -//! budget ([`asap_accuracy::budget`]); `k`'s sizes are those of its -//! `parts`, so a lone denominator sized for the whole target -//! (`ε/(1 − ε) > ε`) fails here; +//! - every output exact: `u`'s values do not reach the output along the +//! path (only a row count does), and there is no term; +//! - every output meets the target: weight `One`, one share of `q`'s budget +//! ([`asap_accuracy::budget`]); `u`'s sizes are those of its `parts`, so +//! a lone denominator sized for the whole target (`ε/(1 − ε) > ε`) fails +//! here, while the same state read as the numerator meets it; //! - otherwise [`AccuracyWeight::Unusable`], with the check's cause //! ([`UnusableReason`]): e.g. a rank error through any arithmetic, or an //! HLL whose declared distinct bound is outside its contract. //! -//! For each pair `k1, k2` of candidates usable alone, both sized for at -//! least two shares, whose values meet in one plan of `q` (under two inputs -//! of one candidate, under two references of a scalar root, or one below -//! the other), the outputs are built and checked with both approximate; a -//! pair that fails is excluded (`a[q, k1] + a[q, k2] ≤ 1`). A pair with a -//! member sized for one share is excluded by the shares row already. The -//! rows still bound the shares a plan reads by each candidate's `parts`, -//! which the equal split proves for one product or quotient of two inputs; -//! three or more approximate inputs are not built together. +//! For each pair of terms with weight `One`, both producers sized for at least +//! two shares, that one plan of `q` can depend on together (paths from two +//! references of a scalar root, paths leaving each other at two inputs of +//! one consumer, or one path continuing the other), the outputs are built +//! and checked with both approximate; a pair that fails is excluded. A pair +//! with a producer sized for one share is excluded by the shares row +//! already. The rows still bound the shares a plan depends on by each +//! producer's `parts`, which the equal split proves for one product or +//! quotient of two inputs; three or more approximate values are not built +//! together. use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use asap_cost::{CostInputs, NodeUsage, ProducerOutput}; @@ -152,12 +162,13 @@ pub(crate) struct MilpInputs { /// Per ingestion-time shareable candidate whose context reads a /// candidate of an `I_r`: the index of that `I_r` in `contexts`. pub(crate) ingestion_context_of_shareable: BTreeMap, - /// Per query: `w(q, k)` for each approximate candidate `k` whose values - /// reach its output. - pub(crate) accuracy_weights: Vec>, - /// Per query: the pairs `(k1, k2)`, `k1 < k2`, of candidates usable alone - /// that it may not read together. - pub(crate) accuracy_excluded_pairs: Vec>, + /// Per query: its row (K) terms, one per path of edges from an + /// approximate producer whose values reach its output. + pub(crate) accuracy_terms: Vec>, + /// Per query: the pairs `(t1, t2)`, `t1 < t2`, of indices into its + /// `accuracy_terms`, each with weight `One` alone, that it may not depend on + /// together. + pub(crate) accuracy_excluded_pairs: Vec>, } /// One candidate's data for the MILP. @@ -300,16 +311,43 @@ pub(crate) struct DominatedCandidate { pub(crate) reason: String, } -/// `w(q, k)`: one share of `q`'s budget, or unusable. +/// One term of row (K) for query `q`: approximate +/// producer `u`'s values reaching `q`'s output along one path of edges +/// `(u, c, i)`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct AccuracyTerm { + /// The output reference of `q` the path starts at. + pub(crate) output_reference: usize, + /// The path's candidates, from an output candidate of + /// `output_reference` down to `u`, the last. + pub(crate) candidates: Vec, + /// Per edge of the path, top down: the input `i` at which consumer + /// `candidates[d]` reads `candidates[d + 1]`. + pub(crate) inputs: Vec, + /// The term's accuracy weight along this path: one budget share or + /// unusable, not row (K)'s constant λ (interim, deleted by #1207 step 6). + pub(crate) weight: AccuracyWeight, +} + +impl AccuracyTerm { + /// `u`, the approximate producer. + pub(crate) fn producer(&self) -> PhysicalCandidateId { + *self.candidates.last().expect("a path has a candidate") + } +} + +/// The accuracy weight of one term: one share of `q`'s budget, or +/// unusable. Interim: #1207 step 6 replaces it with row (K)'s constant +/// λ_{q,c,i}. #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] pub(crate) enum AccuracyWeight { One, - /// `∞`: `k` is unusable for `q`. + /// `∞`: `u`'s values are unusable for `q` along this path. Unusable(UnusableReason), } -/// Why a position's weight is `∞`: the check of `q`'s outputs with `k` -/// approximate failed. +/// Why a term's weight is `Unusable`: the check of `q`'s outputs with `u`'s values +/// approximate along the term's path failed. #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] pub(crate) enum UnusableReason { /// Some output has no guarantee @@ -571,10 +609,10 @@ pub(crate) fn prepare_milp_inputs_with( &refresh_windows, usable, ); - let AccuracyPositions { - weights: accuracy_weights, + let AccuracyTerms { + terms: accuracy_terms, excluded_pairs: accuracy_excluded_pairs, - } = accuracy_positions::accuracy_positions( + } = accuracy_terms::accuracy_terms( physical, &candidate_data, &producer_candidates_per_input, @@ -593,7 +631,7 @@ pub(crate) fn prepare_milp_inputs_with( contexts, context_of_shareable, ingestion_context_of_shareable, - accuracy_weights, + accuracy_terms, accuracy_excluded_pairs, } } @@ -2155,9 +2193,9 @@ pub(crate) fn prune_dominated_candidates( pruned } -mod accuracy_positions; +mod accuracy_terms; -use accuracy_positions::AccuracyPositions; +use accuracy_terms::AccuracyTerms; #[cfg(test)] mod tests; diff --git a/crates/plan-selection/src/groups/accuracy_positions.rs b/crates/plan-selection/src/groups/accuracy_positions.rs deleted file mode 100644 index fa81af61f..000000000 --- a/crates/plan-selection/src/groups/accuracy_positions.rs +++ /dev/null @@ -1,777 +0,0 @@ -//! Accuracy weights and pair exclusions per query (module documentation of -//! [`super`], "Accuracy weights"), each from `asap-accuracy`'s composition -//! and Stage 3's own check ([`guarantee_check`]) of an output built over -//! the candidates. -//! -//! For query `q` and a set `S` of approximate candidates (one candidate, or -//! a pair), the outputs `q` can have with exactly `S` approximate are built -//! from the candidates' operator nodes: a candidate in `S` is its own node; -//! a candidate with no member of `S` below it stands in as its node with an -//! exact guarantee; any other candidate is its node rebuilt over each -//! combination of its producers' outputs, with its guarantee composed again -//! ([`composed_guarantee`]), as the realizer recomposes a selected plan. -//! Outputs over which readers compose the same guarantee are kept once -//! ([`reader_key`]). Each output is then checked as -//! the selected plan is: [`guarantee_check`] with the deployment's model. -//! `S` is usable for `q` only when every output passes: a candidate that -//! fails through one reader is unusable through every reader (the largest -//! weight counts, as before). The candidates of one operator node (other -//! timings or materializations) build the same node, so they are checked -//! once, together. What a check builds is kept for the later checks -//! ([`BuiltBefore`]). -use std::cell::OnceCell; -use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; -use std::hash::{Hash, Hasher}; -use std::rc::Rc; - -use asap_accuracy::propagation::{composed_guarantee, root_guarantee, with_guarantees}; -use asap_types::ir::properties::AccuracyGuarantee; -use asap_types::ir::{OperatorNode, OperatorResultKind, QueryRoot}; -use asap_types::stages::{output_references, PhysicalASAPMemoCandidates, PhysicalCandidateId}; -use asap_types::workload::{DataWorkload, QueryWorkloadEntry}; - -use super::{AccuracyWeight, CandidateData, UnusableReason}; -use crate::checks::guarantee_check; -use crate::report::RejectionCause; -use crate::SelectionModels; - -/// What the accuracy rows (#1095 §2.0 row 9) are built from. -#[derive(Debug, Clone, Default)] -pub(crate) struct AccuracyPositions { - /// Per query: `w(q, k)` for each approximate candidate `k` whose values - /// reach `q`'s output. - pub(crate) weights: Vec>, - /// Per query: the pairs `(k1, k2)`, `k1 < k2`, each usable alone, whose - /// outputs miss `q`'s target when both are approximate: - /// `a[q, k1] + a[q, k2] ≤ 1`. - pub(crate) excluded_pairs: Vec>, -} - -/// The outcome of one check of a query with a set of approximate candidates. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum Outcome { - /// Every output is exact: no value of the set reaches the output. - Exact, - /// Every output meets the target. - Meets, - Fails(UnusableReason), -} - -/// [`AccuracyPositions`] of `physical`'s queries over the candidate links -/// left (`producer_candidates_per_input`, `output_candidates`). -pub(crate) fn accuracy_positions( - physical: &PhysicalASAPMemoCandidates, - candidate_data: &[CandidateData], - producer_candidates_per_input: &[Vec>], - output_candidates: &[Vec], - entries: &[QueryWorkloadEntry], - data: &DataWorkload, - models: &SelectionModels<'_>, -) -> AccuracyPositions { - let below = approximate_below(candidate_data, producer_candidates_per_input); - let output_refs = output_references(&physical.query_roots); - let exact_stand_ins: Vec>> = (0..physical.candidates.len()) - .map(|_| OnceCell::new()) - .collect(); - let mut built_before = BuiltBefore::default(); - let mut positions = AccuracyPositions::default(); - for (query, root) in physical.query_roots.iter().enumerate() { - let references: Vec = (0..output_refs.len()) - .filter(|&reference| output_refs[reference].0 == query) - .collect(); - let reachable: BTreeSet = references - .iter() - .flat_map(|&reference| &output_candidates[reference]) - .flat_map(|class| below[class.0].iter().copied()) - .collect(); - let Some(entry) = entries.get(query) else { - positions.weights.push(BTreeMap::new()); - positions.excluded_pairs.push(Vec::new()); - continue; - }; - let checker = Checker { - physical, - exact_stand_ins: &exact_stand_ins, - producer_candidates_per_input, - output_candidates, - below: &below, - query, - root, - references: &references, - entry, - data, - models, - }; - // Candidates of one operator node (other timings or - // materializations) build the same node: checked once, together. - let mut groups: BTreeMap> = BTreeMap::new(); - for &candidate in &reachable { - groups - .entry(physical.candidates[candidate].operator_node.0) - .or_default() - .push(candidate); - } - let mut weights = BTreeMap::new(); - let mut usable: Vec<&[usize]> = Vec::new(); - for group in groups.values() { - let weight = match checker.check(group, &mut built_before) { - Outcome::Exact => continue, - Outcome::Meets => { - usable.push(group); - AccuracyWeight::One - } - Outcome::Fails(reason) => AccuracyWeight::Unusable(reason), - }; - for &candidate in group { - weights.insert(PhysicalCandidateId(candidate), weight); - } - } - let usable_candidates: Vec = usable - .iter() - .flat_map(|group| group.iter().copied()) - .collect(); - let group_of: HashMap = usable - .iter() - .enumerate() - .flat_map(|(index, group)| group.iter().map(move |&candidate| (candidate, index))) - .collect(); - let group_pairs: BTreeSet<(usize, usize)> = pairs_that_meet( - &usable_candidates, - &references, - physical, - producer_candidates_per_input, - output_candidates, - &below, - ) - .into_iter() - .map(|(first, second)| (group_of[&first], group_of[&second])) - .filter(|(first, second)| first != second) - .map(|(first, second)| (first.min(second), first.max(second))) - // A pair with a member sized for fewer than two shares is never - // read together: the shares row already excludes it. - .filter(|&(first, second)| { - [first, second] - .iter() - .all(|&group| candidate_data[usable[group][0]].accuracy_budget_parts >= 2) - }) - .collect(); - let mut excluded = Vec::new(); - for (first, second) in group_pairs { - let both: Vec = usable[first] - .iter() - .chain(usable[second]) - .copied() - .collect(); - if let Outcome::Fails(_) = checker.check(&both, &mut built_before) { - for &one in usable[first] { - for &other in usable[second] { - excluded.push(( - PhysicalCandidateId(one.min(other)), - PhysicalCandidateId(one.max(other)), - )); - } - } - } - } - excluded.sort_unstable(); - positions.weights.push(weights); - positions.excluded_pairs.push(excluded); - } - positions -} - -/// Per candidate, the approximate candidates at or below it through the -/// links, ascending. A fixpoint, so cycles of candidate links are covered. -fn approximate_below( - candidate_data: &[CandidateData], - producer_candidates_per_input: &[Vec>], -) -> Vec> { - let mut below: Vec> = candidate_data - .iter() - .enumerate() - .map(|(class, data)| match data.approximate { - true => BTreeSet::from([class]), - false => BTreeSet::new(), - }) - .collect(); - loop { - let mut changed = false; - for class in 0..below.len() { - let mut grown = below[class].clone(); - for producer in producer_candidates_per_input[class].iter().flatten() { - grown.extend(below[producer.0].iter().copied()); - } - if grown.len() != below[class].len() { - below[class] = grown; - changed = true; - } - } - if !changed { - return below; - } - } -} - -/// The pairs of `usable` (ascending, `first < second`) one plan of the -/// query can read together where their values meet: under two different -/// inputs of one candidate, under two output references of a scalar root, -/// or one below the other. -fn pairs_that_meet( - usable: &[usize], - references: &[usize], - physical: &PhysicalASAPMemoCandidates, - producer_candidates_per_input: &[Vec>], - output_candidates: &[Vec], - below: &[BTreeSet], -) -> BTreeSet<(usize, usize)> { - let usable: BTreeSet = usable.iter().copied().collect(); - let mut pairs = BTreeSet::new(); - let mut add_across = |sides: &[BTreeSet]| { - for (index, left) in sides.iter().enumerate() { - for right in &sides[index + 1..] { - for &a in left { - for &b in right { - if a != b { - pairs.insert((a.min(b), a.max(b))); - } - } - } - } - } - }; - let side = |producers: &[PhysicalCandidateId]| -> BTreeSet { - producers - .iter() - .flat_map(|producer| below[producer.0].iter().copied()) - .filter(|candidate| usable.contains(candidate)) - .collect() - }; - for producers_per_input in producer_candidates_per_input - .iter() - .take(physical.candidates.len()) - { - if producers_per_input.len() < 2 { - continue; - } - let sides: Vec> = producers_per_input - .iter() - .map(|producers| side(producers)) - .collect(); - add_across(&sides); - } - let sides: Vec> = references - .iter() - .map(|&reference| side(&output_candidates[reference])) - .collect(); - add_across(&sides); - for &upper in &usable { - for &lower in &below[upper] { - if lower != upper && usable.contains(&lower) { - pairs.insert((lower.min(upper), lower.max(upper))); - } - } - } - pairs -} - -/// One query's checks. -struct Checker<'a> { - physical: &'a PhysicalASAPMemoCandidates, - /// Per candidate, its node with an exact guarantee, built once. - exact_stand_ins: &'a [OnceCell>], - producer_candidates_per_input: &'a [Vec>], - output_candidates: &'a [Vec], - below: &'a [BTreeSet], - query: usize, - root: &'a QueryRoot, - references: &'a [usize], - entry: &'a QueryWorkloadEntry, - data: &'a DataWorkload, - models: &'a SelectionModels<'a>, -} - -/// The outputs one candidate is built as, or why they are not checked. -type Built = Result>, UnusableReason>; - -/// What the checks built, kept across checks and queries. A candidate's -/// outputs depend only on the members of the approximate set at or below -/// it, so a pair check rebuilds only the candidates above both members -/// and reuses the rest from the single checks. Outputs built across a -/// cycle of links depend on where the walk entered the cycle and are not -/// kept. -#[derive(Default)] -struct BuiltBefore { - /// Per (candidate, approximate members at or below it, ascending): its - /// outputs. - outputs: HashMap<(usize, Vec), Built>, - /// The [`reader_key`]s of every node built so far. - keys: Keys, -} - -#[cfg(test)] -thread_local! { - /// Set by a test to check with no kept builds: every check builds - /// from scratch ([`BuiltBefore`] cleared) and builds every output - /// candidate (no exact stand-in skipped). - pub(crate) static FRESH_BUILDS: std::cell::Cell = const { std::cell::Cell::new(false) }; -} - -/// One check's walk. -#[derive(Default)] -struct Walk { - /// Per candidate built in this check. - built: HashMap, - /// The candidates being built. - visiting: Vec, - /// How many reads of a candidate still being built the walk has made. - cycle_reads: usize, -} - -/// A candidate's outputs in one check, and whether they were built across -/// a cycle of links (then they are not kept in [`BuiltBefore`]). -struct BuiltInWalk { - outputs: Built, - across_cycle: bool, -} - -impl Checker<'_> { - /// Checks the query's outputs with exactly `approximate` approximate. - /// `before` holds what earlier checks built, for every query. - fn check(&self, approximate: &[usize], before: &mut BuiltBefore) -> Outcome { - #[cfg(test)] - let fresh = FRESH_BUILDS.with(std::cell::Cell::get); - #[cfg(not(test))] - let fresh = false; - if fresh { - *before = BuiltBefore::default(); - } - let mut walk = Walk::default(); - let mut per_reference: Vec>> = Vec::new(); - for &reference in self.references { - let mut outputs: Vec> = Vec::new(); - // Exact stand-ins that are not summary states are the same - // for readers ([`reader_key`]): only the first is kept. - let mut exact_value_stand_in = false; - for &class in &self.output_candidates[reference] { - let reaches = approximate - .iter() - .any(|member| self.below[class.0].contains(member)); - if !fresh - && !reaches - && self.physical.physical_node(class).result_kind != OperatorResultKind::State - && std::mem::replace(&mut exact_value_stand_in, true) - { - continue; - } - match self.built(class.0, approximate, &mut walk, before) { - Ok(built) => outputs.extend(built), - Err(reason) => return Outcome::Fails(reason), - } - } - per_reference.push(distinct_for_readers(outputs, &mut before.keys)); - } - let roots = self.roots(&per_reference); - let mut outcome = Outcome::Exact; - for root in roots { - if root_guarantee(&root).is_some_and(|g| g.is_exact()) { - continue; - } - match guarantee_check( - self.query + 1, - &root, - &self.entry.requirements.accuracy, - self.data, - self.models, - ) { - Ok(_) => outcome = Outcome::Meets, - Err(failure) => { - return Outcome::Fails(match failure.cause { - RejectionCause::NoAccuracyGuarantee => UnusableReason::NoAccuracyGuarantee, - _ => UnusableReason::AccuracyTargetMissed, - }) - } - } - } - outcome - } - - /// The query roots over each combination of `per_reference`'s outputs. - fn roots(&self, per_reference: &[Vec>]) -> Vec { - combinations(per_reference) - .into_iter() - .map(|chosen| match self.root { - QueryRoot::Operator(_) => QueryRoot::Operator(chosen[0].clone()), - QueryRoot::Scalar(expr) => { - let mut next = chosen.into_iter(); - QueryRoot::Scalar( - expr.map_operator_refs(&mut |_| { - next.next().expect("one output per reference") - }), - ) - } - }) - .collect() - } - - /// `class`'s node with an exact guarantee, built once. - fn exact_stand_in(&self, class: usize) -> Rc { - self.exact_stand_ins[class] - .get_or_init(|| { - let node = self.physical.physical_node(PhysicalCandidateId(class)); - Rc::new( - (**node) - .clone() - .with_guarantee(Some(AccuracyGuarantee::exact("an exact stand-in"))), - ) - }) - .clone() - } - - /// The outputs `class` is built as with `approximate` approximate - /// (module documentation). `walk.visiting` holds the candidates being - /// built, so a cycle of links builds nothing. - fn built( - &self, - class: usize, - approximate: &[usize], - walk: &mut Walk, - before: &mut BuiltBefore, - ) -> Built { - if let Some(done) = walk.built.get(&class) { - walk.cycle_reads += usize::from(done.across_cycle); - return done.outputs.clone(); - } - if walk.visiting.contains(&class) { - walk.cycle_reads += 1; - return Ok(Vec::new()); - } - let reaches = |candidate: usize| { - approximate - .iter() - .any(|member| self.below[candidate].contains(member)) - }; - if !reaches(class) { - return Ok(vec![self.exact_stand_in(class)]); - } - let mut members: Vec = approximate - .iter() - .copied() - .filter(|member| self.below[class].contains(member)) - .collect(); - members.sort_unstable(); - let key = (class, members); - if let Some(done) = before.outputs.get(&key) { - walk.built.insert( - class, - BuiltInWalk { - outputs: done.clone(), - across_cycle: false, - }, - ); - return done.clone(); - } - let cycle_reads = walk.cycle_reads; - let node = self.physical.physical_node(PhysicalCandidateId(class)); - let producers = &self.producer_candidates_per_input[class]; - let below_inputs = producers - .iter() - .flatten() - .any(|producer| reaches(producer.0)); - let built = if !below_inputs { - // `class` is in `approximate`, over exact inputs: its node with - // every guarantee below it composed, as the realizer does. - with_guarantees(node, &mut HashMap::new()) - .map(|node| vec![node]) - .map_err(|_| UnusableReason::NotChecked) - } else { - walk.visiting.push(class); - let mut per_input = Vec::with_capacity(producers.len()); - let mut failed = None; - for input in producers { - let mut outputs = Vec::new(); - for producer in input.iter().filter(|producer| reaches(producer.0)) { - match self.built(producer.0, approximate, walk, before) { - Ok(built) => outputs.extend(built), - Err(reason) => failed = Some(reason), - } - } - // One exact stand-in for the producers with no member below: - // one with no approximate candidate below, when there is one. - let exact = input - .iter() - .filter(|producer| !reaches(producer.0)) - .min_by_key(|producer| !self.below[producer.0].is_empty()); - if let Some(exact) = exact { - outputs.push(self.exact_stand_in(exact.0)); - } - per_input.push(distinct_for_readers(outputs, &mut before.keys)); - } - walk.visiting.pop(); - match failed { - Some(reason) => Err(reason), - None => rebuilt(node, &per_input, &mut before.keys), - } - }; - let across_cycle = walk.cycle_reads != cycle_reads; - if !across_cycle { - before.outputs.insert(key, built.clone()); - } - walk.built.insert( - class, - BuiltInWalk { - outputs: built.clone(), - across_cycle, - }, - ); - built - } -} - -/// `node` rebuilt over each combination of `per_input`'s outputs, with its -/// timing and materialization and its guarantee composed again; outputs the -/// same for their readers kept once. -fn rebuilt(node: &Rc, per_input: &[Vec>], keys: &mut Keys) -> Built { - let combinations = combinations(per_input); - let mut outputs = Vec::with_capacity(combinations.len()); - for chosen in combinations { - let mut next = chosen.into_iter(); - let rebuilt = node - .with_new_children(|_| next.next().expect("one output per input")) - .map_err(|_| UnusableReason::NotChecked)?; - let guarantee = composed_guarantee(&rebuilt); - outputs.push(Rc::new( - rebuilt - .with_guarantee(guarantee) - .with_timing(node.timing) - .with_materialization(node.materialization), - )); - } - Ok(distinct_for_readers(outputs, keys)) -} - -/// Every choice of one entry per list, in order. One empty list gives none. -fn combinations(lists: &[Vec]) -> Vec> { - let mut combinations: Vec> = vec![Vec::new()]; - for list in lists { - combinations = combinations - .into_iter() - .flat_map(|prefix| { - list.iter().map(move |item| { - let mut next = prefix.clone(); - next.push(item.clone()); - next - }) - }) - .collect(); - } - combinations -} - -/// `outputs` with outputs the same for their readers kept once (the -/// first): equal [`reader_key`]s. -fn distinct_for_readers(outputs: Vec>, keys: &mut Keys) -> Vec> { - let mut seen = HashSet::new(); - outputs - .into_iter() - .filter(|output| seen.insert(reader_key(output, keys))) - .collect() -} - -/// Per node (by address): its [`reader_key`], and whether binding -/// statistics leaves its guarantee as it is; the node is held so its -/// address is not reused during the check. -type Keys = HashMap<*const OperatorNode, (u64, bool, Rc)>; - -/// A key equal for outputs over which readers compose the same guarantee: -/// an exact node as one key (an exact summary state, a stand-in, by its -/// address); a node with a guarantee (not a summary state) by its bound and failure -/// probability (binding keeps a node's guarantee and ignores its inputs; -/// provenance is not read); a node with none that binding cannot give one -/// as one key; any other node (a summary state, whose readers read its -/// inputs' guarantees, or a node with none that binding may still give -/// one) by its operator kind and its inputs' keys. -fn reader_key(node: &Rc, keys: &mut Keys) -> u64 { - key_and_settled(node, keys).0 -} - -/// [`reader_key`], and whether binding statistics leaves `node`'s -/// guarantee as it is: a guarantee with no unknown statistic, or none over -/// inputs that are all settled. -fn key_and_settled(node: &Rc, keys: &mut Keys) -> (u64, bool) { - if let Some(&(key, settled, _)) = keys.get(&Rc::as_ptr(node)) { - return (key, settled); - } - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - // Exact whatever its inputs (an exact stand-in among them). - if node - .guarantee - .as_ref() - .is_some_and(|guarantee| guarantee.is_exact()) - { - "exact".hash(&mut hasher); - if node.result_kind == OperatorResultKind::State { - Rc::as_ptr(node).hash(&mut hasher); - } - let known = (hasher.finish(), true); - keys.insert(Rc::as_ptr(node), (known.0, known.1, node.clone())); - return known; - } - let spec = node.guarantee.as_ref().map(|guarantee| &guarantee.spec); - format!("{spec:?}").hash(&mut hasher); - let state = node.result_kind == OperatorResultKind::State; - let known = match &node.guarantee { - Some(guarantee) if !state => ( - hasher.finish(), - guarantee.spec.error.evaluate().is_some() - && guarantee.spec.failure_probability.evaluate().is_some(), - ), - _ => { - let children: Vec<(u64, bool)> = node - .children() - .into_iter() - .map(|child| key_and_settled(child, keys)) - .collect(); - let settled = node.guarantee.is_none() && children.iter().all(|(_, settled)| *settled); - if !(settled && !state) { - node.operator.kind_name().hash(&mut hasher); - state.hash(&mut hasher); - for (key, _) in &children { - key.hash(&mut hasher); - } - } - (hasher.finish(), settled) - } - }; - keys.insert(Rc::as_ptr(node), (known.0, known.1, node.clone())); - known -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::groups::tests::{candidate, catalog, reader_of, relative, root, values_node}; - use asap_logical_asap_optimization::test_support::entry_with; - use asap_types::ir::properties::AccuracyTarget; - use asap_types::workload::{QueryRecurrence, RepeatedDemand, RepetitionInterval}; - - /// The depth of each output of `built`. - fn depths(built: &Built) -> Vec { - fn depth(node: &OperatorNode) -> usize { - 1 + node - .children() - .into_iter() - .map(|child| depth(child)) - .max() - .unwrap_or(0) - } - built - .as_ref() - .expect("built") - .iter() - .map(|node| depth(node)) - .collect() - } - - // A two-candidate link cycle above an approximate member: `A` reads - // the leaf or `B`; `B` reads `A` or a second leaf. Walking from the - // top enters the cycle at `A`, so `B` is built there with `A` cut off - // (a cycle read), over the second leaf only. Such outputs are not - // kept: a later walk that starts at `B` builds it in full, over `A`, - // as with no kept builds, for one member and for a pair. - #[test] - fn outputs_built_across_a_cycle_are_not_kept() { - let approximate_leaf = |value: f64, error: f64| { - Rc::new( - (*values_node(value)) - .clone() - .with_guarantee(Some(relative(error, 0.01))), - ) - }; - let leaf = approximate_leaf(1.0, 0.01); - let other_leaf = approximate_leaf(2.0, 0.02); - let a = reader_of(&leaf); - let b = reader_of(&leaf); - let top = reader_of(&a); - let physical = catalog( - vec![(leaf, 0), (a, 1), (b, 2), (top, 3), (other_leaf, 4)], - vec![ - candidate(0, &[], false), - candidate(1, &[&[0, 2]], false), - candidate(2, &[&[1, 4]], false), - candidate(3, &[&[1]], true), - candidate(4, &[], false), - ], - vec![root(3)], - ); - let producers: Vec>> = physical - .candidates - .iter() - .map(|candidate| { - candidate - .producer_candidates_per_input - .iter() - .map(|producers| producers.iter().copied().collect()) - .collect() - }) - .collect(); - // Leaves 0 and 4 stand for approximate candidates. - let below: Vec> = vec![ - BTreeSet::from([0]), - BTreeSet::from([0, 4]), - BTreeSet::from([0, 4]), - BTreeSet::from([0, 4]), - BTreeSet::from([4]), - ]; - let exact_stand_ins: Vec>> = (0..physical.candidates.len()) - .map(|_| OnceCell::new()) - .collect(); - let entry = entry_with( - AccuracyTarget::max_error(0.01).failure_probability(0.01), - QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(RepetitionInterval(60_000))), - ); - let data = DataWorkload::default(); - let cost_inputs = asap_cost::CostInputs::default(); - let models = SelectionModels { - accuracy: &asap_accuracy::BuiltinAccuracyModel, - evidence: &asap_accuracy::NoAccuracyEvidence, - cost_inputs: &cost_inputs, - }; - let output_candidates = vec![vec![PhysicalCandidateId(3)]]; - let checker = Checker { - physical: &physical, - exact_stand_ins: &exact_stand_ins, - producer_candidates_per_input: &producers, - output_candidates: &output_candidates, - below: &below, - query: 0, - root: &physical.query_roots[0], - references: &[0], - entry: &entry, - data: &data, - models: &models, - }; - for approximate in [&[0][..], &[0, 4]] { - let mut kept = BuiltBefore::default(); - // From the top: `B` is built with `A` cut off, over the second - // leaf alone (depth 2). - let mut walk = Walk::default(); - checker - .built(3, approximate, &mut walk, &mut kept) - .expect("built"); - assert!(walk.cycle_reads > 0); - assert!(walk.built[&2].across_cycle); - assert_eq!(depths(&walk.built[&2].outputs)[0], 2, "{approximate:?}"); - // From `B`: in full, its first output reads `A` over the leaf - // (depth 3), with kept builds as with none. - let from_b = checker.built(2, approximate, &mut Walk::default(), &mut kept); - let fresh = checker.built( - 2, - approximate, - &mut Walk::default(), - &mut BuiltBefore::default(), - ); - assert_eq!(depths(&from_b), depths(&fresh), "{approximate:?}"); - assert_eq!(depths(&fresh)[0], 3, "{approximate:?}"); - } - } -} diff --git a/crates/plan-selection/src/groups/accuracy_terms.rs b/crates/plan-selection/src/groups/accuracy_terms.rs new file mode 100644 index 000000000..1c3fb4fc0 --- /dev/null +++ b/crates/plan-selection/src/groups/accuracy_terms.rs @@ -0,0 +1,639 @@ +//! Row (K)'s terms per query (module documentation of [`super`], "Row (K) +//! terms"), each checked by `asap-accuracy`'s composition and Stage 3's own +//! check ([`guarantee_check`]) of an output built over the candidates. +//! +//! For query `q`, a term is one path of edges `(u, c, i)` from an +//! approximate producer `u` up to an output candidate of `q`. A set `S` of +//! terms (one term, or a pair) is checked on the output `q` has with `u`'s +//! values approximate along exactly the paths of `S`: each consumer on a +//! path is its node rebuilt over the path's producer at the path's input +//! and an exact stand-in at every other input, with its guarantee composed +//! again ([`composed_guarantee`]), as the realizer recomposes a selected +//! plan; an output reference no path of `S` starts at reads exact +//! stand-ins of its output candidates. Each output is then checked as the +//! selected plan is: [`guarantee_check`] with the deployment's model. +//! +//! Terms whose paths go through the same operator nodes (candidates of one +//! operator node at other timings or materializations build the same node) +//! are checked once, together. What a check builds is kept for the later +//! checks ([`BuiltBefore`]): a consumer's outputs depend only on the paths +//! below it. +use std::collections::{BTreeSet, HashMap, HashSet}; +use std::hash::{Hash, Hasher}; +use std::rc::Rc; + +use asap_accuracy::propagation::{composed_guarantee, root_guarantee, with_guarantees}; +use asap_types::ir::properties::AccuracyGuarantee; +use asap_types::ir::{OperatorNode, OperatorResultKind, QueryRoot}; +use asap_types::stages::{output_references, PhysicalASAPMemoCandidates, PhysicalCandidateId}; +use asap_types::workload::{DataWorkload, QueryWorkloadEntry}; + +use super::{AccuracyTerm, AccuracyWeight, CandidateData, UnusableReason}; +use crate::checks::guarantee_check; +use crate::report::RejectionCause; +use crate::SelectionModels; + +/// What row (K) is built from. +#[derive(Debug, Clone, Default)] +pub(crate) struct AccuracyTerms { + /// Per query: its terms, one per path of edges from an approximate + /// producer `u` whose values reach its output. + pub(crate) terms: Vec>, + /// Per query: the pairs `(t1, t2)`, `t1 < t2`, of indices into its + /// terms, each with weight `One` alone, whose outputs miss `q`'s target when + /// both are approximate. + pub(crate) excluded_pairs: Vec>, +} + +/// The outcome of one check of a query with a set of terms approximate. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Outcome { + /// Every output is exact: no value of the set reaches the output. + Exact, + /// Every output meets the target. + Meets, + Fails(UnusableReason), +} + +/// [`AccuracyTerms`] of `physical`'s queries over the candidate links left +/// (`producer_candidates_per_input`, `output_candidates`). +pub(crate) fn accuracy_terms( + physical: &PhysicalASAPMemoCandidates, + candidate_data: &[CandidateData], + producer_candidates_per_input: &[Vec>], + output_candidates: &[Vec], + entries: &[QueryWorkloadEntry], + data: &DataWorkload, + models: &SelectionModels<'_>, +) -> AccuracyTerms { + let below = approximate_below(candidate_data, producer_candidates_per_input); + let output_refs = output_references(&physical.query_roots); + let exact_stand_ins: Vec>> = + (0..physical.candidates.len()) + .map(|_| std::cell::OnceCell::new()) + .collect(); + let mut built_before = BuiltBefore::default(); + let mut result = AccuracyTerms::default(); + for (query, root) in physical.query_roots.iter().enumerate() { + let references: Vec = (0..output_refs.len()) + .filter(|&reference| output_refs[reference].0 == query) + .collect(); + let Some(entry) = entries.get(query) else { + result.terms.push(Vec::new()); + result.excluded_pairs.push(Vec::new()); + continue; + }; + let checker = Checker { + physical, + exact_stand_ins: &exact_stand_ins, + producer_candidates_per_input, + output_candidates, + below: &below, + query, + root, + references: &references, + entry, + data, + models, + }; + let paths = paths_to_approximate_producers( + candidate_data, + producer_candidates_per_input, + output_candidates, + &below, + &references, + ); + // Paths through the same operator nodes build the same outputs: + // checked once, together. + let mut outcomes: HashMap = HashMap::new(); + let mut terms = Vec::new(); + for path in paths { + let key = checker.node_path(&path); + let outcome = *outcomes + .entry(key) + .or_insert_with(|| checker.check(&[&path], &mut built_before)); + let weight = match outcome { + Outcome::Exact => continue, + Outcome::Meets => AccuracyWeight::One, + Outcome::Fails(reason) => AccuracyWeight::Unusable(reason), + }; + terms.push(AccuracyTerm { weight, ..path }); + } + let mut pair_outcomes: HashMap<(NodePath, NodePath), Outcome> = HashMap::new(); + let mut excluded = Vec::new(); + for first in 0..terms.len() { + for second in first + 1..terms.len() { + let (one, other) = (&terms[first], &terms[second]); + // A pair with a term sized for fewer than two shares is + // never read together: the shares row already excludes it. + if [one, other].iter().any(|term| { + term.weight != AccuracyWeight::One + || candidate_data[term.producer().0].accuracy_budget_parts < 2 + }) || !read_together(one, other) + { + continue; + } + let key = (checker.node_path(one), checker.node_path(other)); + let outcome = *pair_outcomes + .entry(key) + .or_insert_with(|| checker.check(&[one, other], &mut built_before)); + if let Outcome::Fails(_) = outcome { + excluded.push((first, second)); + } + } + } + result.terms.push(terms); + result.excluded_pairs.push(excluded); + } + result +} + +/// Per candidate, the approximate candidates at or below it through the +/// links, ascending. A fixpoint, so cycles of candidate links are covered. +fn approximate_below( + candidate_data: &[CandidateData], + producer_candidates_per_input: &[Vec>], +) -> Vec> { + let mut below: Vec> = candidate_data + .iter() + .enumerate() + .map(|(class, data)| match data.approximate { + true => BTreeSet::from([class]), + false => BTreeSet::new(), + }) + .collect(); + loop { + let mut changed = false; + for class in 0..below.len() { + let mut grown = below[class].clone(); + for producer in producer_candidates_per_input[class].iter().flatten() { + grown.extend(below[producer.0].iter().copied()); + } + if grown.len() != below[class].len() { + below[class] = grown; + changed = true; + } + } + if !changed { + return below; + } + } +} + +/// Every path of edges from an output candidate of `references` down to an +/// approximate producer, no candidate twice on one path (links can close a +/// cycle), in order of reference, then depth first by input and producer. +/// Each is returned as a term with weight `One`, set by the caller. +fn paths_to_approximate_producers( + candidate_data: &[CandidateData], + producer_candidates_per_input: &[Vec>], + output_candidates: &[Vec], + below: &[BTreeSet], + references: &[usize], +) -> Vec { + struct Walk<'a> { + candidate_data: &'a [CandidateData], + producer_candidates_per_input: &'a [Vec>], + below: &'a [BTreeSet], + reference: usize, + found: Vec, + } + impl Walk<'_> { + fn walk(&mut self, candidates: &mut Vec, inputs: &mut Vec) { + let candidate = *candidates.last().expect("a path has a candidate"); + if self.candidate_data[candidate.0].approximate { + self.found.push(AccuracyTerm { + output_reference: self.reference, + candidates: candidates.clone(), + inputs: inputs.clone(), + weight: AccuracyWeight::One, + }); + } + for (input, producers) in self.producer_candidates_per_input[candidate.0] + .iter() + .enumerate() + { + for &producer in producers { + if self.below[producer.0].is_empty() || candidates.contains(&producer) { + continue; + } + candidates.push(producer); + inputs.push(input); + self.walk(candidates, inputs); + candidates.pop(); + inputs.pop(); + } + } + } + } + let mut walk = Walk { + candidate_data, + producer_candidates_per_input, + below, + reference: 0, + found: Vec::new(), + }; + for &reference in references { + walk.reference = reference; + for &output in &output_candidates[reference] { + if !below[output.0].is_empty() { + walk.walk(&mut vec![output], &mut Vec::new()); + } + } + } + walk.found +} + +/// Whether one plan of the query can read the values of both terms: their +/// paths start at different output references, or leave each other at +/// two different inputs of one consumer, or one path continues the other. +/// Two paths through different candidates of one output reference or of +/// one consumer's input are never in one plan. +pub(super) fn read_together(one: &AccuracyTerm, other: &AccuracyTerm) -> bool { + if one.output_reference != other.output_reference { + return true; + } + for depth in 0.. { + match (one.candidates.get(depth), other.candidates.get(depth)) { + (Some(a), Some(b)) if a == b => {} + (Some(_), Some(_)) => return false, + _ => return true, + } + match (one.inputs.get(depth), other.inputs.get(depth)) { + (Some(a), Some(b)) if a != b => return true, + _ => {} + } + } + unreachable!() +} + +/// A term's path by operator nodes: `(output reference, operator node per +/// candidate, inputs)`. +type NodePath = (usize, Vec, Vec); + +/// One query's checks. +struct Checker<'a> { + physical: &'a PhysicalASAPMemoCandidates, + /// Per candidate, its node with an exact guarantee, built once. + exact_stand_ins: &'a [std::cell::OnceCell>], + producer_candidates_per_input: &'a [Vec>], + output_candidates: &'a [Vec], + below: &'a [BTreeSet], + query: usize, + root: &'a QueryRoot, + references: &'a [usize], + entry: &'a QueryWorkloadEntry, + data: &'a DataWorkload, + models: &'a SelectionModels<'a>, +} + +/// The outputs one consumer is built as, or why they are not checked. +type Built = Result>, UnusableReason>; + +/// The tails of the paths below one candidate: per path, its candidates +/// and inputs from that candidate down. +type Tails = Vec<(Vec, Vec)>; + +/// What the checks built, kept across checks and queries: per candidate +/// and the tails of the approximate paths below it, its outputs. +#[derive(Default)] +struct BuiltBefore { + outputs: HashMap, + /// The [`reader_key`]s of every node built so far. + keys: Keys, +} + +#[cfg(test)] +thread_local! { + /// Set by a test to check with no kept builds: every check builds + /// from scratch ([`BuiltBefore`] cleared) and builds every output + /// candidate of a reference no path starts at (no exact stand-in + /// skipped). + pub(crate) static FRESH_BUILDS: std::cell::Cell = const { std::cell::Cell::new(false) }; +} + +impl Checker<'_> { + fn node_path(&self, term: &AccuracyTerm) -> NodePath { + ( + term.output_reference, + term.candidates + .iter() + .map(|candidate| self.physical.candidates[candidate.0].operator_node.0) + .collect(), + term.inputs.clone(), + ) + } + + /// Checks the query's outputs with the values of `terms`' producers + /// approximate along their paths. `before` holds what earlier checks + /// built, for every query. + fn check(&self, terms: &[&AccuracyTerm], before: &mut BuiltBefore) -> Outcome { + #[cfg(test)] + let fresh = FRESH_BUILDS.with(std::cell::Cell::get); + #[cfg(not(test))] + let fresh = false; + if fresh { + *before = BuiltBefore::default(); + } + let mut per_reference: Vec>> = Vec::new(); + for &reference in self.references { + let tails: Tails = terms + .iter() + .filter(|term| term.output_reference == reference) + .map(|term| (term.candidates.clone(), term.inputs.clone())) + .collect(); + let outputs = if tails.is_empty() { + // Exact stand-ins that are not summary states are the same + // for readers ([`reader_key`]): only the first is kept. + let mut outputs = Vec::new(); + let mut exact_value_stand_in = false; + for &class in &self.output_candidates[reference] { + if !fresh + && self.physical.physical_node(class).result_kind + != OperatorResultKind::State + && std::mem::replace(&mut exact_value_stand_in, true) + { + continue; + } + outputs.push(self.exact_stand_in(class.0)); + } + outputs + } else { + match self.built(tails, before) { + Ok(built) => built, + Err(reason) => return Outcome::Fails(reason), + } + }; + per_reference.push(distinct_for_readers(outputs, &mut before.keys)); + } + let roots = self.roots(&per_reference); + let mut outcome = Outcome::Exact; + for root in roots { + if root_guarantee(&root).is_some_and(|g| g.is_exact()) { + continue; + } + match guarantee_check( + self.query + 1, + &root, + &self.entry.requirements.accuracy, + self.data, + self.models, + ) { + Ok(_) => outcome = Outcome::Meets, + Err(failure) => { + return Outcome::Fails(match failure.cause { + RejectionCause::NoAccuracyGuarantee => UnusableReason::NoAccuracyGuarantee, + _ => UnusableReason::AccuracyTargetMissed, + }) + } + } + } + outcome + } + + /// The query roots over each combination of `per_reference`'s outputs. + fn roots(&self, per_reference: &[Vec>]) -> Vec { + combinations(per_reference) + .into_iter() + .map(|chosen| match self.root { + QueryRoot::Operator(_) => QueryRoot::Operator(chosen[0].clone()), + QueryRoot::Scalar(expr) => { + let mut next = chosen.into_iter(); + QueryRoot::Scalar( + expr.map_operator_refs(&mut |_| { + next.next().expect("one output per reference") + }), + ) + } + }) + .collect() + } + + /// `class`'s node with an exact guarantee, built once. + fn exact_stand_in(&self, class: usize) -> Rc { + self.exact_stand_ins[class] + .get_or_init(|| { + let node = self.physical.physical_node(PhysicalCandidateId(class)); + Rc::new( + (**node) + .clone() + .with_guarantee(Some(AccuracyGuarantee::exact("an exact stand-in"))), + ) + }) + .clone() + } + + /// The outputs of the candidate every tail of `tails` starts at, with + /// the values of each tail's last candidate approximate along it. + fn built(&self, tails: Tails, before: &mut BuiltBefore) -> Built { + if let Some(done) = before.outputs.get(&tails) { + return done.clone(); + } + let class = tails[0].0[0]; + let node = self.physical.physical_node(class); + let continuing: Vec<&(Vec, Vec)> = tails + .iter() + .filter(|(candidates, _)| candidates.len() > 1) + .collect(); + let built = if continuing.is_empty() { + // `class` is the approximate producer of every tail, over exact + // inputs: its node with every guarantee below it composed, as + // the realizer does. + with_guarantees(node, &mut HashMap::new()) + .map(|node| vec![node]) + .map_err(|_| UnusableReason::NotChecked) + } else { + let producers = &self.producer_candidates_per_input[class.0]; + let mut per_input = Vec::with_capacity(producers.len()); + let mut failed = None; + for (input, producers) in producers.iter().enumerate() { + let below: Tails = continuing + .iter() + .filter(|(_, inputs)| inputs[0] == input) + .map(|(candidates, inputs)| (candidates[1..].to_vec(), inputs[1..].to_vec())) + .collect(); + if below.is_empty() { + // One exact stand-in: a producer with no approximate + // candidate below, when there is one. + let exact = producers + .iter() + .min_by_key(|producer| !self.below[producer.0].is_empty()); + per_input.push( + exact + .map(|exact| self.exact_stand_in(exact.0)) + .into_iter() + .collect(), + ); + } else { + match self.built(below, before) { + Ok(built) => per_input.push(built), + Err(reason) => { + failed = Some(reason); + per_input.push(Vec::new()); + } + } + } + } + match failed { + Some(reason) => Err(reason), + None => rebuilt(node, &per_input, &mut before.keys), + } + }; + before.outputs.insert(tails, built.clone()); + built + } +} + +/// `node` rebuilt over each combination of `per_input`'s outputs, with its +/// timing and materialization and its guarantee composed again; outputs the +/// same for their readers kept once. +fn rebuilt(node: &Rc, per_input: &[Vec>], keys: &mut Keys) -> Built { + let combinations = combinations(per_input); + let mut outputs = Vec::with_capacity(combinations.len()); + for chosen in combinations { + let mut next = chosen.into_iter(); + let rebuilt = node + .with_new_children(|_| next.next().expect("one output per input")) + .map_err(|_| UnusableReason::NotChecked)?; + let guarantee = composed_guarantee(&rebuilt); + outputs.push(Rc::new( + rebuilt + .with_guarantee(guarantee) + .with_timing(node.timing) + .with_materialization(node.materialization), + )); + } + Ok(distinct_for_readers(outputs, keys)) +} + +/// Every choice of one entry per list, in order. One empty list gives none. +fn combinations(lists: &[Vec]) -> Vec> { + let mut combinations: Vec> = vec![Vec::new()]; + for list in lists { + combinations = combinations + .into_iter() + .flat_map(|prefix| { + list.iter().map(move |item| { + let mut next = prefix.clone(); + next.push(item.clone()); + next + }) + }) + .collect(); + } + combinations +} + +/// `outputs` with outputs the same for their readers kept once (the +/// first): equal [`reader_key`]s. +fn distinct_for_readers(outputs: Vec>, keys: &mut Keys) -> Vec> { + let mut seen = HashSet::new(); + outputs + .into_iter() + .filter(|output| seen.insert(reader_key(output, keys))) + .collect() +} + +/// Per node (by address): its [`reader_key`], and whether binding +/// statistics leaves its guarantee as it is; the node is held so its +/// address is not reused during the check. +type Keys = HashMap<*const OperatorNode, (u64, bool, Rc)>; + +/// A key equal for outputs over which readers compose the same guarantee: +/// an exact node as one key (an exact summary state, a stand-in, by its +/// address); a node with a guarantee (not a summary state) by its bound and failure +/// probability (binding keeps a node's guarantee and ignores its inputs; +/// provenance is not read); a node with none that binding cannot give one +/// as one key; any other node (a summary state, whose readers read its +/// inputs' guarantees, or a node with none that binding may still give +/// one) by its operator kind and its inputs' keys. +fn reader_key(node: &Rc, keys: &mut Keys) -> u64 { + key_and_settled(node, keys).0 +} + +/// [`reader_key`], and whether binding statistics leaves `node`'s +/// guarantee as it is: a guarantee with no unknown statistic, or none over +/// inputs that are all settled. +fn key_and_settled(node: &Rc, keys: &mut Keys) -> (u64, bool) { + if let Some(&(key, settled, _)) = keys.get(&Rc::as_ptr(node)) { + return (key, settled); + } + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + // Exact whatever its inputs (an exact stand-in among them). + if node + .guarantee + .as_ref() + .is_some_and(|guarantee| guarantee.is_exact()) + { + "exact".hash(&mut hasher); + if node.result_kind == OperatorResultKind::State { + Rc::as_ptr(node).hash(&mut hasher); + } + let known = (hasher.finish(), true); + keys.insert(Rc::as_ptr(node), (known.0, known.1, node.clone())); + return known; + } + let spec = node.guarantee.as_ref().map(|guarantee| &guarantee.spec); + format!("{spec:?}").hash(&mut hasher); + let state = node.result_kind == OperatorResultKind::State; + let known = match &node.guarantee { + Some(guarantee) if !state => ( + hasher.finish(), + guarantee.spec.error.evaluate().is_some() + && guarantee.spec.failure_probability.evaluate().is_some(), + ), + _ => { + let children: Vec<(u64, bool)> = node + .children() + .into_iter() + .map(|child| key_and_settled(child, keys)) + .collect(); + let settled = node.guarantee.is_none() && children.iter().all(|(_, settled)| *settled); + if !(settled && !state) { + node.operator.kind_name().hash(&mut hasher); + state.hash(&mut hasher); + for (key, _) in &children { + key.hash(&mut hasher); + } + } + (hasher.finish(), settled) + } + }; + keys.insert(Rc::as_ptr(node), (known.0, known.1, node.clone())); + known +} + +#[cfg(test)] +mod tests { + use super::*; + + fn term(reference: usize, candidates: &[usize], inputs: &[usize]) -> AccuracyTerm { + AccuracyTerm { + output_reference: reference, + candidates: candidates.iter().map(|&c| PhysicalCandidateId(c)).collect(), + inputs: inputs.to_vec(), + weight: AccuracyWeight::One, + } + } + + // One state `u = 9` read by both operands of a quotient `0` through a + // shared merge `5`: `0` reads estimate `1` at input 0 and estimate `2` + // at input 1, each over `5`. The two paths leave each other at `0`'s + // two inputs, so one plan depends on both; two paths through different + // candidates of one input, or of one output reference, are never in + // one plan; a path ending on the other's path is. + #[test] + fn terms_read_together_where_their_paths_can_meet() { + let numerator = term(0, &[0, 1, 5, 9], &[0, 0, 0]); + let denominator = term(0, &[0, 2, 5, 9], &[1, 0, 0]); + assert!(read_together(&numerator, &denominator)); + let other_estimate = term(0, &[0, 3, 5, 9], &[0, 0, 0]); + assert!(!read_together(&numerator, &other_estimate)); + let other_output = term(0, &[4, 5, 9], &[0, 0]); + assert!(!read_together(&numerator, &other_output)); + let above = term(0, &[0, 1, 5], &[0, 0]); + assert!(read_together(&numerator, &above)); + let second_reference = term(1, &[4, 5, 9], &[0, 0]); + assert!(read_together(&numerator, &second_reference)); + } +} diff --git a/crates/plan-selection/src/groups/tests.rs b/crates/plan-selection/src/groups/tests.rs index f9c2ce4ee..b344c1769 100644 --- a/crates/plan-selection/src/groups/tests.rs +++ b/crates/plan-selection/src/groups/tests.rs @@ -661,32 +661,30 @@ fn two_queries_share_a_pane_chain_through_the_real_pipeline() { ); } } - for weights in &inputs.accuracy_weights { - assert!(!weights.is_empty()); + for terms in &inputs.accuracy_terms { + assert!(!terms.is_empty()); assert!( - weights - .values() - .all(|&weight| weight == AccuracyWeight::One), - "{weights:?}" + terms.iter().all(|term| term.weight == AccuracyWeight::One), + "{terms:?}" ); } } // --------------------------------------------------------------------------- -// Accuracy weights +// Row (K) terms // --------------------------------------------------------------------------- -/// The sketch algorithm and `parts` of each approximate candidate of -/// `physical`, with its weight for query 0 (`None`: no weight). -fn sketch_weights( +/// The sketch algorithm and `parts` of the producer of each row (K) term of +/// query 0, with the term's weight. +fn sketch_terms( physical: &PhysicalASAPMemoCandidates, inputs: &MilpInputs, -) -> Vec<(SketchAlgorithm, usize, Option)> { - (0..physical.candidates.len()) - .filter(|&class| inputs.candidate_data[class].approximate) - .map(|class| { - let id = PhysicalCandidateId(class); - let algorithm = match physical.physical_node(id).asap() { +) -> Vec<(SketchAlgorithm, usize, AccuracyWeight)> { + inputs.accuracy_terms[0] + .iter() + .map(|term| { + let producer = term.producer(); + let algorithm = match physical.physical_node(producer).asap() { Some( ASAPOp::SummaryAgg { family: FieldDataType::Sketch(kind, _), @@ -697,23 +695,23 @@ fn sketch_weights( .. }, ) => *kind.algorithm(), - _ => unreachable!("an approximate candidate is a sketch"), + _ => unreachable!("an approximate producer is a sketch"), }; ( algorithm, - inputs.candidate_data[class].accuracy_budget_parts, - inputs.accuracy_weights[0].get(&id).copied(), + inputs.candidate_data[producer.0].accuracy_budget_parts, + term.weight, ) }) .collect() } -// `q0.5 / q0.9` through the real stages, weights from the composition: a -// KLL rank error has no rule through a quotient, so Pass 1 prunes every KLL -// candidate (#1129); a DDSketch relative -// error composes, but a lone denominator sized for the whole target -// (`parts = 1`) misses it (`ε/(1 − ε) > ε`) while one sized for two parts -// meets it. +// `q0.5 / q0.9` through the real stages, weights from the composition: a KLL +// rank error has no rule through a quotient, so Pass 1 prunes every KLL +// candidate (#1129); a DDSketch relative error composes, but a lone +// denominator sized for the whole target (`parts = 1`) misses it +// (`ε/(1 − ε) > ε`) while a numerator sized so meets it, and one sized for +// two parts meets it in either operand. #[test] fn ratio_weights_come_from_the_composition() { let stages = through_stages_1_and_2(&[ @@ -725,36 +723,34 @@ fn ratio_weights_come_from_the_composition() { &stages.data, &CostInputs::default(), ); - let weights = sketch_weights(&stages.physical, &inputs); - let of = |algorithm: SketchAlgorithm, parts: usize| -> BTreeSet> { - weights + let terms = sketch_terms(&stages.physical, &inputs); + let of = |algorithm: SketchAlgorithm, parts: usize| -> BTreeSet { + terms .iter() .filter(|(a, p, _)| *a == algorithm && *p == parts) .map(|(_, _, weight)| *weight) .collect() }; - let missed = Some(AccuracyWeight::Unusable( - UnusableReason::AccuracyTargetMissed, - )); + let missed = AccuracyWeight::Unusable(UnusableReason::AccuracyTargetMissed); // Pass 1 prunes every KLL: its rank error composes no guarantee // through the quotient. for parts in [1, 2] { assert_eq!( of(SketchAlgorithm::Kll, parts), BTreeSet::new(), - "{weights:?}" + "{terms:?}" ); } // At one part: numerators meet the target, denominators miss it. assert_eq!( of(SketchAlgorithm::DDSketch, 1), - BTreeSet::from([Some(AccuracyWeight::One), missed]), - "{weights:?}" + BTreeSet::from([AccuracyWeight::One, missed]), + "{terms:?}" ); assert_eq!( of(SketchAlgorithm::DDSketch, 2), - BTreeSet::from([Some(AccuracyWeight::One)]), - "{weights:?}" + BTreeSet::from([AccuracyWeight::One]), + "{terms:?}" ); // Two DDSketches at two parts compose: no pair is excluded. assert!(inputs.accuracy_excluded_pairs[0].is_empty()); @@ -764,7 +760,8 @@ fn ratio_weights_come_from_the_composition() { // alone. One DDSketch pane chain sized for two shares of the budget serves // both quotient operands and both lone quantiles, as the old Stage 3 // engine planned it: the check accepts the quotient over it (`(α + α)/(1 − α) ≤ ε` at -// `α = 1 − (1 + ε)^{−1/2}`), so it weighs 1 for every query. A chain sized +// `α = 1 − (1 + ε)^{−1/2}`), so every term of every query over it has +// weight `One`. A chain sized // for the whole budget (`α = ε`) never serves the quotient's denominator. #[test] fn one_pane_chain_serves_a_ratio_and_its_operands() { @@ -793,15 +790,19 @@ fn one_pane_chain_serves_a_ratio_and_its_operands() { }, _ => None, }; - let weight = |query: usize, class: usize| { - inputs.accuracy_weights[query] - .get(&PhysicalCandidateId(class)) - .copied() + // the weight of every term of `query` whose producer is `class`. + let lambdas = |query: usize, class: usize| -> BTreeSet { + inputs.accuracy_terms[query] + .iter() + .filter(|term| term.producer() == PhysicalCandidateId(class)) + .map(|term| term.weight) + .collect() }; + let one = BTreeSet::from([AccuracyWeight::One]); let two_shares = 1.0 - 1.01f64.powf(-0.5); let shared: Vec = (0..stages.physical.candidates.len()) .filter(|&class| alpha(class).is_some()) - .filter(|&class| (0..3).all(|query| weight(query, class) == Some(AccuracyWeight::One))) + .filter(|&class| (0..3).all(|query| lambdas(query, class) == one)) .collect(); assert!(!shared.is_empty()); for &class in &shared { @@ -811,22 +812,20 @@ fn one_pane_chain_serves_a_ratio_and_its_operands() { alpha(class) ); } - // A chain sized for the whole budget weighs 1 for the ratio only where + // A chain sized for the whole budget has weight `One` for the ratio only where // its values are the numerator's alone, so it never also serves both // lone quantiles. for class in (0..stages.physical.candidates.len()).filter(|&class| alpha(class) == Some(0.01)) { - assert!( - !(0..3).all(|query| weight(query, class) == Some(AccuracyWeight::One)), - "{class}" - ); + assert!(!(0..3).all(|query| lambdas(query, class) == one), "{class}"); } } // The checks keep what they build for later checks and skip repeated // exact stand-ins: on RangeHeavy T8+T5's shape (a quotient and both -// operands alone), on a quotient of three quantiles (where pairs are -// excluded) and on a ratio with one approximate operand, the weights and -// excluded pairs equal those with no kept builds and no skipped stand-in. +// operands alone), on a quotient of three quantiles (where pairs of terms +// are checked) and on a ratio with one approximate operand, the row (K) terms +// and excluded pairs equal those with no kept builds and no skipped +// stand-in. #[test] fn kept_builds_give_the_weights_of_fresh_builds() { kept_builds_match_fresh_builds(&[ @@ -837,12 +836,24 @@ fn kept_builds_give_the_weights_of_fresh_builds() { let three = kept_builds_match_fresh_builds(&[ "quantile_over_time(0.9, m[5m]) / quantile_over_time(0.5, m[5m]) / quantile_over_time(0.1, m[5m])", ]); - assert!(!three.accuracy_excluded_pairs[0].is_empty()); + // Pair checks run: a state Pass 2 shares among the three operands is + // the producer of terms with weight `One` under two inputs of one quotient. + // (Each pair of them meets the target; three of them together are + // excluded by the shares row, not by a pair.) + let terms = &three.accuracy_terms[0]; + assert!(terms.iter().enumerate().any(|(index, one)| { + terms[index + 1..].iter().any(|other| { + one.producer() == other.producer() + && one.weight == AccuracyWeight::One + && other.weight == AccuracyWeight::One + && three.candidate_data[one.producer().0].accuracy_budget_parts >= 2 + && accuracy_terms::read_together(one, other) + }) + })); // A scalar ratio of two output references whose approximate values sit // under one operand only: the other reference's output candidates - // reach no member, and its first exact stand-in must still be built, - // or the ratio has no root to check and every approximate candidate - // would weigh nothing. + // reach no producer, and its first exact stand-in must still be built, + // or the ratio has no root to check and there would be no term. use asap_logical_asap_optimization::test_support::lower_promql; use asap_types::ir::scalar::{ArithmeticOpKind, ExprSemantics}; let operand = |query: &str| { @@ -858,11 +869,11 @@ fn kept_builds_give_the_weights_of_fresh_builds() { semantics: ExprSemantics::Promql, }); let ratio = kept_builds_match_fresh_builds_of(vec![ratio]); - assert!(!ratio.accuracy_weights[0].is_empty()); + assert!(!ratio.accuracy_terms[0].is_empty()); } -/// Asserts that `queries`' accuracy weights and excluded pairs are the -/// same with kept builds and with none ([`accuracy_positions::FRESH_BUILDS`]); +/// Asserts that `queries`' row (K) terms and excluded pairs are the same +/// with kept builds and with none ([`accuracy_terms::FRESH_BUILDS`]); /// returns the former. fn kept_builds_match_fresh_builds(queries: &[&str]) -> MilpInputs { use asap_logical_asap_optimization::test_support::lower_promql; @@ -878,22 +889,19 @@ fn kept_builds_match_fresh_builds(queries: &[&str]) -> MilpInputs { fn kept_builds_match_fresh_builds_of(roots: Vec) -> MilpInputs { let stages = through_stages_1_and_2_of(roots); let prepare = |fresh: bool| { - accuracy_positions::FRESH_BUILDS.with(|cell| cell.set(fresh)); + accuracy_terms::FRESH_BUILDS.with(|cell| cell.set(fresh)); let inputs = prepare_milp_inputs( &stages.physical, &stages.entries, &stages.data, &CostInputs::default(), ); - accuracy_positions::FRESH_BUILDS.with(|cell| cell.set(false)); + accuracy_terms::FRESH_BUILDS.with(|cell| cell.set(false)); inputs }; let (kept, fresh) = (prepare(false), prepare(true)); - assert!(kept - .accuracy_weights - .iter() - .any(|weights| !weights.is_empty())); - assert_eq!(kept.accuracy_weights, fresh.accuracy_weights); + assert!(kept.accuracy_terms.iter().any(|terms| !terms.is_empty())); + assert_eq!(kept.accuracy_terms, fresh.accuracy_terms); assert_eq!(kept.accuracy_excluded_pairs, fresh.accuracy_excluded_pairs); kept } diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index 77100f7df..0693b26b7 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -248,7 +248,7 @@ mod accuracy_binding; mod checks; mod models; // #1095 §2.0: the group selection MILP's inputs (dominance, execution -// contexts, accuracy weights). +// contexts, row (K) terms). mod groups; // #1095 §2.0: the group selection MILP and the decode of its solution. mod milp; diff --git a/crates/plan-selection/src/milp/model.rs b/crates/plan-selection/src/milp/model.rs index ffd268768..a89385f33 100644 --- a/crates/plan-selection/src/milp/model.rs +++ b/crates/plan-selection/src/milp/model.rs @@ -8,7 +8,8 @@ //! | `p[ctx, c]` | in context `ctx`, candidate `c` gives its result slot (binary): computed there, or read across the boundary (a shareable candidate, or a candidate of an ingestion pipeline `I_r`) | //! | `x[m]` | shareable candidate `m` is selected (binary) | //! | `a[q, m]` | query `q` depends on shareable `m` (continuous, exact at integral binaries) | -//! | `y[q, m, c]` | `AND(a[q, m], p[m, c])`, for reach from above and accuracy (continuous) | +//! | `y[q, m, c]` | `AND(a[q, m], p[m, c])`, for reach from above (continuous) | +//! | `d[q, t]` | `q` depends on row (K) term `t`: `AND` of its path's indicators (continuous) | //! | `L[ctx]`, `T[q, m]`, `K[q]` | latency of a query or of a query-time shareable candidate's run and context; `m`'s latency in `q`'s; compaction (continuous) | //! | `N[q]`, `A[q, parts]` | approximate values `q` depends on; whether it depends on one sized for `parts` shares (continuous) | //! | `g[m, c]` | highest shared materialization: `c` is read in context `m`, which at most one query reaches (binary) | @@ -30,7 +31,7 @@ //! 6. Reach: `a[q, m] ≥ p[q, m]`, `a[q, c] ≥ a[q, m] + p[m, c] − 1`; from //! above, where a reader constraint counts queries. //! 7. Latency (`plan_cost`'s: each query-time run it reaches counts once). -//! 8. Accuracy (A1), per run a query depends on. +//! 8. Accuracy (A1), per row (K) term a query depends on. //! 9. Reader constraints. //! //! A link from a candidate to one of its own result slot is dropped: with @@ -51,7 +52,7 @@ use good_lp::{ SolverModel, Variable, }; -use crate::groups::{AccuracyWeight, ContextKind, MilpInputs, ResultSlot, RunKind}; +use crate::groups::{AccuracyTerm, AccuracyWeight, ContextKind, MilpInputs, ResultSlot, RunKind}; use super::cost_terms::{self, RunTerm}; use super::warm_start; @@ -1582,106 +1583,151 @@ fn latency_rows(milp: &mut SelectionMilp<'_>) -> Result<(), SelectionMilpError> Ok(()) } -/// Row 8: accuracy (A1, revised 2026-10-31), per run `q` depends on: a run -/// unusable for `q` is never depended on, two whose pair is excluded are -/// not depended on together, and the shares depended on are at most each -/// such run's `parts`. +/// Where a term's path is read in the MILP ([`SelectionMilp::term_path`]). +pub(crate) struct TermPath { + /// The binaries that are all 1 exactly when `q` depends on the term: + /// `p[ctx, c]` for each candidate on the path in the context reading it, + /// and the slot link `e` of each multi-slot input on it. + pub(crate) indicators: Vec, + /// `(context, slot, input)` per candidate on the path (`input` is the + /// edge's input below it, `usize::MAX` for `u`). Two terms through the + /// same places are never both depended on (one candidate per slot per + /// context). + pub(crate) places: Vec<(usize, ResultSlot, usize)>, + /// The context `u` is computed in. + pub(crate) producer_context: usize, +} + +impl SelectionMilp<'_> { + /// The indicators and places of `term`'s path for query `query`: from + /// `query`'s context down, crossing into a shareable candidate's + /// context or the `I_r` its reader reads at each candidate not computed + /// where it is read. `None` when no plan reads the path. + pub(crate) fn term_path(&self, query: usize, term: &AccuracyTerm) -> Option { + let inputs = self.inputs; + let mut context = query; + let mut path = TermPath { + indicators: Vec::new(), + places: Vec::new(), + producer_context: query, + }; + for (depth, &candidate) in term.candidates.iter().enumerate() { + path.indicators.push(*self.choice[context].get(&candidate)?); + let run = if self.computes(context, candidate) { + context + } else { + match inputs.kind[candidate.0] { + RunKind::Shareable => self.context_of(candidate)?, + RunKind::Ingestion => self.ingestion_context_read_by(context)?, + RunKind::PerContext => return None, + } + }; + let input = term.inputs.get(depth).copied(); + path.places.push(( + context, + inputs.slot[candidate.0], + input.unwrap_or(usize::MAX), + )); + match input { + Some(input) => { + if let Some(links) = self.slot_links.get(&(run, candidate, input)) { + let below = inputs.slot[term.candidates[depth + 1].0]; + let (_, link) = links.iter().find(|(slot, _)| *slot == below)?; + path.indicators.push(*link); + } + } + None => path.producer_context = run, + } + context = run; + } + Some(path) + } +} + +/// Row (K), as A1 (revised 2026-10-31) over row (K)'s terms: per query `q`, +/// a term whose weight is unusable is never depended on, two terms whose pair +/// is excluded are not depended on together, and the shares depended on +/// are at most each such term's producer's `parts`. `q` depends on a term +/// when every indicator of its path ([`SelectionMilp::term_path`]) is 1. fn accuracy_rows(milp: &mut SelectionMilp<'_>) { let inputs = milp.inputs; milp.builder.scope = Scope::Global; - let approximate_runs: Vec<(usize, PhysicalCandidateId)> = milp - .runs() - .into_iter() - .filter(|&(_, candidate)| inputs.candidate_data[candidate.0].approximate) - .collect(); for query in 0..milp.query_count() { let exact_target = milp.entries[query].requirements.accuracy.is_exact(); - // (candidate, dep, parts) per run `q` may depend on. - let mut counted: Vec<(PhysicalCandidateId, Expression, usize)> = Vec::new(); - // Where each counted run gives its value: (context, slot), `None` - // for a shareable candidate's own run. One value per place. - let mut places: BTreeSet<(usize, Option)> = BTreeSet::new(); - for &(context, candidate) in &approximate_runs { - let Some(weight) = inputs.accuracy_weights[query].get(&candidate).copied() else { + let terms = &inputs.accuracy_terms[query]; + // An approximate producer of `q` is never computed in an `I_r`. + let producers: BTreeSet = + terms.iter().map(AccuracyTerm::producer).collect(); + for (context, candidate) in milp.runs() { + if producers.contains(&candidate) + && matches!(inputs.contexts[context].kind, ContextKind::Ingestion(_)) + { + milp.note_inexact("an approximate candidate of an I_r (forbidden)"); + let p = milp.choice[context][&candidate]; + milp.builder.row("8 accuracy", constraint!(p <= 0.0)); + } + } + // (term, path, parts) per term `q` may depend on. + let mut counted: Vec<(usize, TermPath, usize)> = Vec::new(); + for (index, term) in terms.iter().enumerate() { + let Some(path) = milp.term_path(query, term) else { continue; }; - let dep: Option = match inputs.contexts[context].kind { - ContextKind::Query(q) if q == query => { - milp.choice[context].get(&candidate).map(|&p| p.into()) - } - ContextKind::Query(_) => None, - ContextKind::Shared(own) if own == candidate => { - milp.reach.get(&(query, candidate)).map(|&a| a.into()) - } - ContextKind::Shared(own) => match milp.reach.get(&(query, own)) { - Some(&above) => { - let p = milp.choice[context][&candidate]; - Some( - milp.builder - .and_at_least( - "y (depends on a run in m)", - &[above.into(), p.into()], - ) - .into(), - ) - } - None => None, - }, - ContextKind::Ingestion(_) => { - milp.note_inexact("an approximate candidate of an I_r (forbidden)"); - let p = milp.choice[context][&candidate]; - milp.builder.row("8 accuracy", constraint!(p <= 0.0)); - None - } - }; - let Some(dep) = dep else { + if matches!( + inputs.contexts[path.producer_context].kind, + ContextKind::Ingestion(_) + ) { continue; - }; - let usable = match weight { - _ if exact_target => false, - AccuracyWeight::One => true, - AccuracyWeight::Unusable(_) => false, - }; + } + let usable = !exact_target && term.weight == AccuracyWeight::One; if usable { - places.insert(match inputs.contexts[context].kind { - ContextKind::Shared(own) if own == candidate => (context, None), - _ => (context, Some(inputs.slot[candidate.0])), - }); - counted.push(( - candidate, - dep, - inputs.candidate_data[candidate.0].accuracy_budget_parts, - )); + let parts = inputs.candidate_data[term.producer().0].accuracy_budget_parts; + counted.push((index, path, parts)); } else { - milp.builder.row("8 accuracy", constraint!(dep <= 0.0)); + not_all(milp, &[&path.indicators]); } } - let mut pairs: Vec<(Expression, Expression)> = Vec::new(); for &(first, second) in &inputs.accuracy_excluded_pairs[query] { - for (_, one, _) in counted.iter().filter(|(c, _, _)| *c == first) { - for (_, other, _) in counted.iter().filter(|(c, _, _)| *c == second) { - pairs.push((one.clone(), other.clone())); - } + let one = counted.iter().find(|(index, _, _)| *index == first); + let other = counted.iter().find(|(index, _, _)| *index == second); + if let (Some((_, one, _)), Some((_, other, _))) = (one, other) { + not_all(milp, &[&one.indicators, &other.indicators]); } } - for (one, other) in pairs { - milp.builder - .row("8 accuracy (pair)", constraint!(one + other <= 1.0)); - } - // The most values `q` can depend on together. + // The most terms `q` can depend on together. let most = if tests_loose_big_m() { counted.len() as f64 } else { - places.len() as f64 + counted + .iter() + .map(|(_, path, _)| &path.places) + .collect::>() + .len() as f64 }; if counted.iter().all(|&(_, _, parts)| parts as f64 >= most) { continue; } + let deps: Vec<(Expression, usize)> = counted + .iter() + .map(|(_, path, parts)| { + let dep: Expression = match path.indicators.as_slice() { + [one] => (*one).into(), + many => { + let terms: Vec = + many.iter().map(|&v| Expression::from(v)).collect(); + milp.builder + .and_at_least("d (q depends on a term)", &terms) + .into() + } + }; + (dep, *parts) + }) + .collect(); let shares = milp.builder.continuous("N (approximate values)", None); - let sum: Expression = counted.iter().map(|(_, dep, _)| dep.clone()).sum(); + let sum: Expression = deps.iter().map(|(dep, _)| dep.clone()).sum(); milp.builder.row("8 accuracy", constraint!(shares == sum)); let mut by_parts: BTreeMap> = BTreeMap::new(); - for (_, dep, parts) in &counted { + for (dep, parts) in &deps { if (*parts as f64) < most { by_parts.entry(*parts).or_default().push(dep.clone()); } @@ -1702,6 +1748,20 @@ fn accuracy_rows(milp: &mut SelectionMilp<'_>) { } } +/// The row that the indicators of `paths` (together) are not all 1. +fn not_all(milp: &mut SelectionMilp<'_>, paths: &[&Vec]) { + let mut indicators: Vec = Vec::new(); + for &indicator in paths.iter().flat_map(|path| path.iter()) { + if !indicators.contains(&indicator) { + indicators.push(indicator); + } + } + let count = indicators.len() as f64; + let sum: Expression = indicators.into_iter().map(Expression::from).sum(); + milp.builder + .row("8 accuracy", constraint!(sum <= count - 1.0)); +} + /// One independent part of the model (#1095 §5, 3a): a connected component /// of the graph joining the variables of each row, or all components /// without a query together. Parts share no variable and no row, and the diff --git a/crates/plan-selection/src/milp/model/tests.rs b/crates/plan-selection/src/milp/model/tests.rs index a8ec9f4a0..bae765d42 100644 --- a/crates/plan-selection/src/milp/model/tests.rs +++ b/crates/plan-selection/src/milp/model/tests.rs @@ -960,11 +960,11 @@ fn cost_models_2_and_3_are_unsupported() { } // Accuracy rows on `q0.5 / q0.9` over metric `m` declared non-negative, -// weights from the composition: a candidate unusable for the query (a -// DDSketch denominator sized for the whole target) is never depended on, -// and two at `parts = 2` are depended on together unless their pair is -// excluded. (Pass 1 already prunes KLL, whose rank error composes no -// guarantee through the quotient.) +// row (K) terms from the composition: an unusable term (a DDSketch +// denominator sized for the whole target) is never depended on, and a +// numerator and a denominator term at `parts = 2` are depended on together +// unless their pair is excluded. (Pass 1 already prunes KLL, whose rank +// error composes no guarantee through the quotient.) #[test] fn accuracy_rows_follow_the_composition_on_a_ratio() { let fixture = through_stages_non_negative(&[approximate( @@ -974,19 +974,11 @@ fn accuracy_rows_follow_the_composition_on_a_ratio() { )]); let costs = cost_inputs(); let inputs = prepare_milp_inputs(&fixture.physical, &fixture.entries, &fixture.data, &costs); - let of = |wanted: &dyn Fn(&AccuracyWeight, usize) -> bool| -> Vec { - inputs.accuracy_weights[0] - .iter() - .filter(|(class, weight)| { - wanted(weight, inputs.candidate_data[class.0].accuracy_budget_parts) - }) - .map(|(class, _)| *class) - .collect() - }; - let unusable = of(&|weight, _| matches!(weight, AccuracyWeight::Unusable(_))); - let usable_at_two = of(&|weight, parts| *weight == AccuracyWeight::One && parts == 2); - assert!(!unusable.is_empty() && usable_at_two.len() >= 2); - let solve = |inputs: &MilpInputs, restrict: &dyn Fn(&mut SelectionMilp<'_>)| { + let terms = &inputs.accuracy_terms[0]; + let parts = |term: usize| inputs.candidate_data[terms[term].producer().0].accuracy_budget_parts; + // Solves with the query depending on every term of `all` (every + // indicator of its path is 1) and on at least one term of `any`. + let solve = |inputs: &MilpInputs, all: &[usize], any: &[usize]| { let mut milp = build_selection_milp( &fixture.physical, inputs, @@ -995,47 +987,72 @@ fn accuracy_rows_follow_the_composition_on_a_ratio() { SelectionMilpOptions::default(), ) .unwrap(); - restrict(&mut milp); - solve_selection_milp(milp, 0).map(|_| ()) - }; - // Each candidate's runs: `x[m]`, or `p` in a context computing it. - let runs_of = |milp: &SelectionMilp<'_>, candidates: &[PhysicalCandidateId]| -> Expression { - let mut sum = Expression::from(0.0); - for &candidate in candidates { - if let Some(&selected) = milp.selected.get(&candidate) { - sum += selected; + for &term in all { + let path = milp + .term_path(0, &terms[term]) + .expect("a path the MILP reads"); + for indicator in path.indicators { + milp.builder.row("test", constraint!(indicator >= 1.0)); } - for context in 0..milp.choice.len() { - if milp.computes(context, candidate) { - if let Some(&choice) = milp.choice[context].get(&candidate) { - sum += choice; - } + } + if !any.is_empty() { + let mut sum = Expression::from(0.0); + for &term in any { + let path = milp + .term_path(0, &terms[term]) + .expect("a path the MILP reads"); + // At most each indicator: 1 only when all are. + let and = milp.builder.continuous("test", Some(1.0)); + for indicator in path.indicators { + milp.builder.row("test", constraint!(and <= indicator)); } + sum += and; } + milp.builder.row("test", constraint!(sum >= 1.0)); } - sum + solve_selection_milp(milp, 0).map(|_| ()) + }; + let readable: Vec = { + let milp = build_selection_milp( + &fixture.physical, + &inputs, + &fixture.entries, + &costs, + SelectionMilpOptions::default(), + ) + .unwrap(); + terms + .iter() + .map(|term| milp.term_path(0, term).is_some()) + .collect() }; + let unusable: Vec = (0..terms.len()) + .filter(|&t| matches!(terms[t].weight, AccuracyWeight::Unusable(_)) && readable[t]) + .collect(); + assert!(!unusable.is_empty()); assert_eq!( - solve(&inputs, &|milp| { - let sum = runs_of(milp, &unusable); - milp.builder.row("test", constraint!(sum >= 1.0)); - }) - .err(), + solve(&inputs, &[], &unusable).err(), Some(SelectionMilpError::Infeasible) ); - let two_of = |milp: &mut SelectionMilp<'_>| { - let sum = runs_of(milp, &usable_at_two); - milp.builder.row("test", constraint!(sum >= 2.0)); - }; - assert!(solve(&inputs, &two_of).is_ok()); + // A numerator and a denominator term at two parts, under the two + // inputs of one quotient candidate. + let at_two = |t: usize| terms[t].weight == AccuracyWeight::One && parts(t) == 2 && readable[t]; + let (numerator, denominator) = (0..terms.len()) + .flat_map(|one| (0..terms.len()).map(move |other| (one, other))) + .filter(|&(one, other)| { + at_two(one) + && at_two(other) + && terms[one].candidates[0] == terms[other].candidates[0] + && terms[one].inputs[0] == 0 + && terms[other].inputs[0] == 1 + }) + .find(|&(one, other)| solve(&inputs, &[one, other], &[]).is_ok()) + .expect("a numerator and a denominator at two parts, read together"); let mut excluded = inputs.clone(); - for (index, &first) in usable_at_two.iter().enumerate() { - for &second in &usable_at_two[index + 1..] { - excluded.accuracy_excluded_pairs[0].push((first, second)); - } - } + excluded.accuracy_excluded_pairs[0] + .push((numerator.min(denominator), numerator.max(denominator))); assert_eq!( - solve(&excluded, &two_of).err(), + solve(&excluded, &[numerator, denominator], &[]).err(), Some(SelectionMilpError::Infeasible) ); } diff --git a/crates/plan-selection/src/milp/warm_start.rs b/crates/plan-selection/src/milp/warm_start.rs index b7de6ae37..b8253b80b 100644 --- a/crates/plan-selection/src/milp/warm_start.rs +++ b/crates/plan-selection/src/milp/warm_start.rs @@ -12,9 +12,9 @@ //! first selected; a candidate of an `I_r` is chosen in the `I_r` of the //! reading context's refresh window. //! -//! **Repairs per query.** If the query's plan depends on more approximate -//! values than one of them is sized for, on a pair excluded together, or -//! on a candidate unusable for it, the query is planned again with exact +//! **Repairs per query.** If the query's plan depends on more row (K) +//! terms than one of their producers is sized for, on a pair of terms +//! excluded together, or on a term unusable for it, the query is planned again with exact //! candidates only; if its estimated latency is over its bound, it is //! planned again by latency alone. //! @@ -29,7 +29,7 @@ use asap_cost::execution_demand::evaluation_rate; use asap_types::stages::{output_references, PhysicalCandidateId, ReaderConstraint}; use good_lp::Variable; -use crate::groups::{AccuracyWeight, ContextKind, ResultSlot, RunKind}; +use crate::groups::{AccuracyTerm, AccuracyWeight, ContextKind, ResultSlot, RunKind}; use super::completion::Completion; use super::cost_terms::NodeNumbers; @@ -425,8 +425,15 @@ impl Derivation<'_, '_> { { return false; } - if let Some(AccuracyWeight::Unusable(_)) = - inputs.accuracy_weights[self.query].get(&candidate) + // Unusable when every row (K) term of the query with this + // producer is (an unusable term needs every candidate on its + // path, so the accuracy check below decides the rest). + let mut terms = inputs.accuracy_terms[self.query] + .iter() + .filter(|term| term.producer() == candidate) + .peekable(); + if terms.peek().is_some() + && terms.all(|term| matches!(term.weight, AccuracyWeight::Unusable(_))) { return false; } @@ -719,36 +726,67 @@ fn runs_reached( reached } -/// Whether query `query`'s approximate runs in `plan` meet the A1 rows. +/// Whether query `query`'s row (K) terms `plan` depends on meet the A1 +/// rows. fn meets_accuracy(milp: &SelectionMilp<'_>, plan: &Plan, query: usize) -> bool { let inputs = milp.inputs; - let runs: Vec = runs_reached(milp, plan, query) - .into_iter() - .map(|(_, candidate)| candidate) - .filter(|candidate| inputs.candidate_data[candidate.0].approximate) - .filter(|candidate| inputs.accuracy_weights[query].contains_key(candidate)) + let depended: Vec = (0..inputs.accuracy_terms[query].len()) + .filter(|&index| depends_on(milp, plan, query, &inputs.accuracy_terms[query][index])) .collect(); - if runs.is_empty() { + if depended.is_empty() { return true; } if milp.entries[query].requirements.accuracy.is_exact() { return false; } - let count = runs.len(); - for &candidate in &runs { - if matches!( - inputs.accuracy_weights[query].get(&candidate), - Some(AccuracyWeight::Unusable(_)) - ) { - return false; - } - if inputs.candidate_data[candidate.0].accuracy_budget_parts < count { + let count = depended.len(); + for &index in &depended { + let term = &inputs.accuracy_terms[query][index]; + if term.weight != AccuracyWeight::One + || inputs.candidate_data[term.producer().0].accuracy_budget_parts < count + { return false; } } !inputs.accuracy_excluded_pairs[query] .iter() - .any(|(first, second)| runs.contains(first) && runs.contains(second)) + .any(|(first, second)| depended.contains(first) && depended.contains(second)) +} + +/// Whether `plan` makes query `query` depend on `term`: every candidate on +/// its path is chosen where the path reads it, as +/// [`SelectionMilp::term_path`] reads the path. +fn depends_on(milp: &SelectionMilp<'_>, plan: &Plan, query: usize, term: &AccuracyTerm) -> bool { + let inputs = milp.inputs; + let mut context = query; + for (depth, &candidate) in term.candidates.iter().enumerate() { + if plan.chosen[context].get(&inputs.slot[candidate.0]) != Some(&candidate) { + return false; + } + let run = if milp.computes(context, candidate) { + context + } else { + let run = match inputs.kind[candidate.0] { + RunKind::Shareable => milp.context_of(candidate), + RunKind::Ingestion => milp.ingestion_context_read_by(context), + RunKind::PerContext => None, + }; + let Some(run) = run else { + return false; + }; + run + }; + if let Some(&input) = term.inputs.get(depth) { + if milp.slot_links.contains_key(&(run, candidate, input)) + && plan.slot_reads.get(&(run, candidate, input)) + != Some(&inputs.slot[term.candidates[depth + 1].0]) + { + return false; + } + } + context = run; + } + true } /// Whether query `query`'s estimated latency in `plan` is within its bound. diff --git a/crates/plan-selection/src/report.rs b/crates/plan-selection/src/report.rs index e2617b580..e91bd7b41 100644 --- a/crates/plan-selection/src/report.rs +++ b/crates/plan-selection/src/report.rs @@ -334,7 +334,7 @@ pub enum ErrorSelectingPlan { }, /// The selected plan fails a check the selection MILP encodes /// (accuracy through the budget shares, latency): a bug in the - /// accuracy weights, the sizing rule or a latency term, never a reason + /// row (K) terms, the sizing rule or a latency term, never a reason /// to re-solve. #[error("the selected plan fails a check the MILP encodes ({cause:?}): {reason}")] SelectedPlanFailedCheck {