refactor(stage1-pass1): per-computation candidates over memo groups (#1207 PR 3)

crates/accuracy/src/propagation.rs

@@ -1605,3 +1605,77 @@ pub fn pinned_column(conjunct: &ScalarExpr) -> Option<usize> {
         _ => None,
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::composition::raw_data_samples_sign_statistic;
+    use asap_types::ir::schema::{
+        GroupingStrategy, NonNegativeWeightProof, SketchAlgorithm, SummaryInputExpr,
+    };
+
+    /// A Count-Min estimate has a bound only over weights proven
+    /// non-negative: an open requirement (`RawDataSamplesOfMetric`) is not a proof.
+    #[test]
+    fn count_min_estimate_has_no_bound_over_an_open_requirement() {
+        let family = FieldDataType::Sketch(SketchAlgorithm::Cms, GroupingStrategy::default());
+        let query = SketchStatistic::PointCount {
+            key: ColumnRef::SampleValue,
+            value: None,
+        };
+        let with = |weight_domain| SummaryUpdate {
+            item: Some(SummaryInputExpr::Column(ColumnRef::SampleValue)),
+            weight: SummaryInputExpr::Column(ColumnRef::SampleValue),
+            weight_domain,
+        };
+        assert!(!estimate_has_bound(
+            &family,
+            &with(WeightDomain::RawDataSamplesOfMetric { metric: "m".into() }),
+            &query
+        ));
+        assert!(!estimate_has_bound(
+            &family,
+            &with(WeightDomain::UnknownOrSigned),
+            &query
+        ));
+        assert!(estimate_has_bound(
+            &family,
+            &with(WeightDomain::NonNegative {
+                proof: NonNegativeWeightProof::CounterRawDataSamples
+            }),
+            &query
+        ));
+    }
+
+    /// A Count-Min estimate over samples of `m` gets its bound gated on
+    /// `m`'s sign: unknown until plan selection binds it, the local bound
+    /// once bound to 1.
+    #[test]
+    fn count_min_estimate_over_raw_data_samples_is_gated_on_the_metric() {
+        let family = FieldDataType::Sketch(SketchAlgorithm::Cms, GroupingStrategy::default());
+        let query = SketchStatistic::PointCount {
+            key: ColumnRef::SampleValue,
+            value: None,
+        };
+        let update = SummaryUpdate {
+            item: Some(SummaryInputExpr::Column(ColumnRef::SampleValue)),
+            weight: SummaryInputExpr::Column(ColumnRef::SampleValue),
+            weight_domain: WeightDomain::RawDataSamplesOfMetric { metric: "m".into() },
+        };
+        let gated = estimate_bound_over_raw_data_samples(&family, &update, &query).expect("gated");
+        assert_eq!(gated.spec.error.evaluate(), None);
+        let local = local_guarantee(&family, &query).unwrap();
+        let statistic = raw_data_samples_sign_statistic("m");
+        // Every other statistic (a stream norm) bound alike.
+        let others = |s: &str| Some(if s == statistic { 1.0 } else { 1_000.0 });
+        let bound = gated.bind(&others).spec.error.evaluate();
+        assert!(bound.is_some());
+        assert_eq!(bound, local.bind(&others).spec.error.evaluate());
+        let unconfirmed = gated
+            .bind(&|s| (s != statistic).then_some(1_000.0))
+            .spec
+            .error
+            .evaluate();
+        assert_eq!(unconfirmed, None);
+    }
+}

crates/accuracy/src/test_support.rs

@@ -7,7 +7,7 @@ use std::rc::Rc;
 use asap_types::ir::operator::operator_properties::Reduction;
 use asap_types::ir::properties::guarantee::AccuracyGuarantee;
 use asap_types::ir::properties::AccuracyTarget;
-use asap_types::ir::schema::{FieldDataType, SketchStatistic, SummaryUpdate};
+use asap_types::ir::schema::{FieldDataType, SketchStatistic};
 use asap_types::ir::{OperatorNode, ScalarExpr};
 
 use crate::composition::StreamNorm;
@@ -42,24 +42,6 @@ pub fn estimates_raw_data_samples_of(node: &OperatorNode) -> Option<String> {
     propagation::estimates_raw_data_samples_of(node)
 }
 
-/// `propagation::estimate_has_bound`.
-pub fn estimate_has_bound(
-    family: &FieldDataType,
-    update: &SummaryUpdate,
-    query: &SketchStatistic,
-) -> bool {
-    propagation::estimate_has_bound(family, update, query)
-}
-
-/// `propagation::estimate_bound_over_raw_data_samples`.
-pub fn estimate_bound_over_raw_data_samples(
-    family: &FieldDataType,
-    update: &SummaryUpdate,
-    query: &SketchStatistic,
-) -> Option<AccuracyGuarantee> {
-    propagation::estimate_bound_over_raw_data_samples(family, update, query)
-}
-
 /// `propagation::source_distinct_guarantee`.
 pub fn source_distinct_guarantee(
     family: &FieldDataType,
@@ -74,14 +56,6 @@ pub fn satisfies(guarantee: &AccuracyGuarantee, target: &AccuracyTarget) -> bool
     crate::satisfies(guarantee, target)
 }
 
-/// `crate::estimators::local_guarantee`.
-pub fn local_guarantee(
-    family: &FieldDataType,
-    query: &SketchStatistic,
-) -> Option<AccuracyGuarantee> {
-    crate::estimators::local_guarantee(family, query)
-}
-
 /// `crate::estimators::cms::cms_depth`.
 pub fn cms_depth(delta: f64) -> u32 {
     crate::estimators::cms::cms_depth(delta)

crates/devtools/Cargo.toml

@@ -37,8 +37,6 @@ tokio = { version = "1", features = ["rt", "macros", "rt-multi-thread"] }
 [dev-dependencies]
 asap-types = { path = "../types", features = ["test-support"] }
 asap-stage2-physical-optimization = { path = "../stage2-physical-optimization" }
-# Pass 1's target list (`per_computation_candidates`) for the tests.
-asap-stage1-pass1-per-computation-candidates = { path = "../stage1-pass1-per-computation-candidates", features = ["test-support"] }
 
 
 [[test]]

crates/devtools/src/bin/sketch_coverage.rs

@@ -4,7 +4,7 @@
 // corpus list exactly, so the two reports are directly comparable) with an
 // *approximate* `AccuracyTarget`, runs Pass 1 of logical ASAP-aware
 // optimization (`pass1_memo`, every
-// summary supported, no accuracy pruning) over each query, and
+// summary supported) over each query, and
 // reports which summary candidates Pass 1 offers for it, plus the fraction of
 // lowered queries with at least one.
 //
@@ -14,7 +14,6 @@
 // every corpus lowers with. Without an approximate target, Pass 1 offers no
 // summary candidate.
 
-use asap_accuracy::{IgnoreAccuracyGuarantees, NoAccuracyEvidence};
 use asap_devtools::corpora::{
     bgp_catalog, dqc_catalog, netflow_catalog, parse_sample_interval, promql_lines, sql_statements,
 };
@@ -25,7 +24,7 @@ use asap_stage1_pass1_per_computation_candidates::pass1_memo;
 use asap_types::ir::properties::AccuracyTarget;
 use asap_types::ir::schema::FieldDataType;
 use asap_types::ir::{ASAPOp, Operator, OperatorNode, QueryRoot};
-use asap_types::workload::{DataWorkload, SqlDialect};
+use asap_types::workload::SqlDialect;
 use std::collections::BTreeSet;
 use std::rc::Rc;
 
@@ -54,7 +53,6 @@ fn analyze_corpus(
     name: &'static str,
     roots: Vec<(String, Rc<OperatorNode>)>,
     failed: usize,
-    accuracy: &AccuracyTarget,
 ) -> CorpusCoverage {
     let lowered = roots.len();
     let mut rejected = 0;
@@ -64,14 +62,7 @@ fn analyze_corpus(
             rejected += 1;
             continue;
         };
-        let Ok(pass1) = pass1_memo(
-            memo,
-            std::slice::from_ref(accuracy),
-            &IgnoreAccuracyGuarantees,
-            &NoAccuracyEvidence,
-            &DataWorkload::default(),
-            None,
-        ) else {
+        let Ok(pass1) = pass1_memo(memo, None) else {
             rejected += 1;
             continue;
         };
@@ -187,7 +178,7 @@ async fn main() {
                 Err(_) => failed += 1,
             }
         }
-        results.push(analyze_corpus(name, roots, failed, &accuracy));
+        results.push(analyze_corpus(name, roots, failed));
     }
 
     // ── SQL corpora ──
@@ -222,7 +213,7 @@ async fn main() {
                 Err(_) => failed += 1,
             }
         }
-        results.push(analyze_corpus(name, roots, failed, &accuracy));
+        results.push(analyze_corpus(name, roots, failed));
     }
 
     for r in &results {

crates/devtools/src/candidate_labels.rs

@@ -390,39 +390,6 @@ mod tests {
     use super::*;
     use crate::planner_layering_examples::example1;
     use crate::planning_stages::lower_promql_roots;
-    use asap_stage1_pass1_per_computation_candidates::test_support::{
-        compose_logical_candidate, generate_per_computation_candidates,
-    };
-
-    /// Selected DAG labels expose accumulator state without a legacy choice ID.
-    #[test]
-    fn connected_labels_describe_selected_states() {
-        let workload = example1();
-        let inventory = generate_per_computation_candidates(
-            &asap_stage0_memo::memo_of_queries(&lower_promql_roots(&workload).unwrap()).unwrap(),
-        )
-        .unwrap();
-        let choice = vec![0; inventory.targets.len()];
-        let roots: Vec<_> = compose_logical_candidate(&inventory, &choice).unwrap();
-        let label = connected_candidate_label(&roots);
-        assert!(label.contains("Q1 ") && label.contains("Q2 "));
-        assert!(label.contains("acc"), "{label}");
-    }
-
-    /// A state reused by two query outputs is named once in the sharing suffix.
-    #[test]
-    fn connected_labels_name_actual_shared_state() {
-        let workload = example1();
-        let inventory = generate_per_computation_candidates(
-            &asap_stage0_memo::memo_of_queries(&lower_promql_roots(&workload).unwrap()).unwrap(),
-        )
-        .unwrap();
-        let roots =
-            compose_logical_candidate(&inventory, &vec![0; inventory.targets.len()]).unwrap();
-        let label = connected_candidate_label(&[roots[1].clone(), roots[1].clone()]);
-        assert!(label.contains("shared Sum acc (Q1, Q2)"), "{label}");
-        assert_eq!(label.matches("shared Sum acc").count(), 1);
-    }
 
     /// Raw logical operators have an exact label, rather than an empty choice label.
     #[test]

crates/devtools/src/planning_stages.rs

@@ -7,7 +7,7 @@ use asap_planner::{
     run_planning, PlanSearchSettings, PlanningModels, PlanningReport, PlanningRun, SelectionError,
 };
 use asap_stage0_memo::memo_of_queries;
-use asap_stage1_pass1_per_computation_candidates::{pass1_memo, Pass1Error};
+use asap_stage1_pass1_per_computation_candidates::pass1_memo;
 use asap_stage1_pass2_reuse_candidates::{
     pass2_after_local_window_composition, pass2_local_window_composition,
 };
@@ -288,38 +288,20 @@ pub fn run_stages_until(
         return Ok(stages);
     }
     let run = &mut stages.run;
-    let accuracy_targets: Vec<_> = entries
-        .iter()
-        .map(|entry| entry.requirements.accuracy.clone())
-        .collect();
-    let accuracy = models.accuracy_check_inputs();
     let pass1 = pass1_memo(
         stage0_dag,
-        &accuracy_targets,
-        accuracy.model,
-        accuracy.evidence,
-        data,
         models
             .deployment_capabilities
             .supported_summaries
             .as_deref(),
-    );
+    )
     // As `run_planning`: a query Pass 1 cannot plan is reported before
     // contradictory deployment capabilities.
-    if let Err(Pass1Error::PerComputation(error)) = &pass1 {
-        return Err(PlanningStagesError::Pass1(error.to_string()));
-    }
+    .map_err(|error| PlanningStagesError::Pass1(error.to_string()))?;
     if let Err(error) = models.deployment_capabilities.validate(data) {
         run.selection = Err(SelectionError::InvalidDeploymentCapabilities(error));
         return Ok(stages);
     }
-    let pass1 = match pass1 {
-        Ok(pass1) => pass1,
-        Err(error) => {
-            run.selection = Err(SelectionError::LogicalASAPCandidates(error.to_string()));
-            return Ok(stages);
-        }
-    };
     run.pass1_memo = Some(pass1.clone());
     if until == RunUntil::Pass1 {
         return Ok(stages);
@@ -372,7 +354,7 @@ pub fn run_stages_until(
         &physical,
         entries,
         data,
-        &accuracy,
+        &models.accuracy_check_inputs(),
         models.cost_inputs,
         &realizer,
     ) {

crates/executor/Cargo.toml

@@ -27,8 +27,6 @@ test-support = ["planner-types/test-support"]
 [dev-dependencies]
 asap-executor = { path = ".", features = ["test-support"] }
 rmp-serde = "1"
-asap-stage0-memo = { path = "../stage0-memo" }
-asap-stage1-pass1-per-computation-candidates = { path = "../stage1-pass1-per-computation-candidates", features = ["test-support"] }
 # The planning stages through the facade (a dev-dependency cycle, for
 # integration tests only).
 asap-planner = { path = "../planner" }

crates/executor/tests/common/mod.rs

@@ -21,38 +21,6 @@ pub fn selected_dag(
     Rc::clone(selected)
 }
 
-/// Exercise each local implementation rule directly, keeping the other
-/// computations at their default choice. A bounded solver archive need not
-/// contain every algorithm that executor binding tests need to exercise.
-pub fn computation_candidates(root: Rc<OperatorNode>) -> Vec<Rc<OperatorNode>> {
-    use asap_stage1_pass1_per_computation_candidates::test_support::{
-        compose_logical_candidate, generate_per_computation_candidates,
-    };
-    let inventory = generate_per_computation_candidates(
-        &asap_stage0_memo::memo_of_queries(&[QueryRoot::Operator(root)]).unwrap(),
-    )
-    .expect("Pass 1 accepts the query");
-    let mut choices = vec![vec![0; inventory.targets.len()]];
-    for (index, target) in inventory.targets.iter().enumerate() {
-        for choice in 1..target.choice_count() {
-            let mut selected = vec![0; inventory.targets.len()];
-            selected[index] = choice;
-            choices.push(selected);
-        }
-    }
-    choices
-        .into_iter()
-        .map(|choice| {
-            let mut roots = compose_logical_candidate(&inventory, &choice)
-                .expect("a generated computation candidate composes");
-            match roots.remove(0) {
-                QueryRoot::Operator(root) => root,
-                QueryRoot::Scalar(_) => panic!("one operator root"),
-            }
-        })
-        .collect()
-}
-
 fn run_all_stages(
     root: Rc<OperatorNode>,
     accuracy: planner_types::ir::properties::AccuracyTarget,

crates/integration-tests/Cargo.toml

@@ -10,7 +10,7 @@ asap-types = { path = "../types", features = ["test-support"] }
 asap-frontend-promql = { path = "../frontend-promql" }
 asap-frontend-sql = { path = "../frontend-sql" }
 asap-stage0-memo = { path = "../stage0-memo" }
-asap-stage1-pass1-per-computation-candidates = { path = "../stage1-pass1-per-computation-candidates", features = ["test-support"] }
+asap-stage1-pass1-per-computation-candidates = { path = "../stage1-pass1-per-computation-candidates" }
 asap-stage2-physical-optimization = { path = "../stage2-physical-optimization" }
 asap-accuracy = { path = "../accuracy" }
 asap-cost = { path = "../cost" }