refactor(stage1): summary readouts and inputs as one operator over groups; Memo::add_candidate; one node equality in Pass 1 (#1203 2a-2)

crates/integration-tests/tests/sketch_rollup.rs

@@ -10,6 +10,7 @@ use std::collections::BTreeMap;
 use std::rc::Rc;
 use std::sync::Arc;
 
+use asap_accuracy::sizing::with_summaries_sized_for_failure_probability;
 use asap_executor::{
     physical_planner::compile,
     runtime::{Limits, RunContext, Scope},
@@ -25,7 +26,8 @@ use asap_types::ir::properties::{AccuracyMetric, AccuracyTarget};
 use asap_types::ir::schema::{
     DataType, ExactKind, ExactParams, Field, FieldDataType, Schema, SketchAlgorithm,
 };
-use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode};
+use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, QueryRoot};
+use asap_types::workload::DataWorkload;
 use futures::StreamExt;
 
 const REGIONS: [&str; 3] = ["us", "eu", "ap"];
@@ -224,13 +226,32 @@ fn rolled_up(coarse: &OperatorNode) -> Rc<OperatorNode> {
 /// rollup from `(region, job)`: the family, the direct root and the
 /// rolled-up root. Explicit rule fixtures keep this execution test independent
 /// of the search archive and its tighter, potentially much larger capacities.
+/// Each candidate's summaries are sized for `accuracy`'s failure
+/// probability, as Pass 1 sizes a memo candidate
+/// ([`with_summaries_sized_for_failure_probability`]); composition alone
+/// sizes them for the calibrated one.
 async fn direct_and_rolled_up(
     sql: &str,
     accuracy: AccuracyTarget,
 ) -> Vec<(SketchAlgorithm, Rc<OperatorNode>, Rc<OperatorNode>)> {
     let root = lower_sql(sql, &catalog(), accuracy.clone()).await.unwrap();
     let mut out = Vec::new();
     for candidate in explicit_local_candidates(root) {
+        let candidate = match accuracy.failure_probability {
+            Some(failure_probability) => {
+                match &with_summaries_sized_for_failure_probability(
+                    &[QueryRoot::Operator(candidate)],
+                    failure_probability,
+                    &DataWorkload::default(),
+                )
+                .unwrap()[..]
+                {
+                    [QueryRoot::Operator(sized)] => Rc::clone(sized),
+                    _ => unreachable!("one operator root in, one out"),
+                }
+            }
+            None => candidate,
+        };
         let Some(build) = build(&candidate) else {
             continue;
         };
@@ -279,14 +300,18 @@ async fn distinct_counts_rolled_up_equal_the_direct_build() {
 
 /// p99 latency per job. DDSketch's bucket counts add, so the merged state
 /// is the direct build's and so is its answer. KLL's compactions are
-/// randomized, so its rolled-up answer differs, but stays within its rank
-/// bound of the exact p99, as the direct build's does.
+/// randomized (the executor's KLL seeds its coin per run), so its rolled-up
+/// answer differs, but stays within its rank bound of the exact p99, as the
+/// direct build's does. Each answer may miss the bound with the sizing's
+/// failure probability: at the calibrated 0.01 (`k = 52`) each of the eight
+/// KLL answers checked here may miss it with probability up to 0.01, so the
+/// KLL is sized for 1e-6 (`k = 100`).
 #[tokio::test]
 async fn quantiles_rolled_up_keep_the_sketch_bound() {
     const EPSILON: f64 = 0.05;
     for metric in [AccuracyMetric::Rank, AccuracyMetric::Relative] {
         let accuracy = AccuracyTarget::max_error(EPSILON)
-            .failure_probability(0.01)
+            .failure_probability(1e-6)
             .accuracy_metric(metric);
         let pairs = direct_and_rolled_up(
             "SELECT job, approx_percentile_cont(latency, 0.99) AS value FROM http GROUP BY job",

crates/logical-asap-optimization/src/lib.rs

@@ -64,6 +64,7 @@ mod added_groups;
 /// The deployment's capabilities: supported summaries and what it retains.
 mod deployment_capabilities;
 mod find_or_add_group;
+mod operator_over_groups;
 mod own_candidate_sizing;
 mod pass1;
 mod pass2;

crates/logical-asap-optimization/src/operator_over_groups.rs

@@ -0,0 +1,95 @@
+//! One operator over groups (#1203): a candidate a rule composes or
+//! rewrites as a tree of operators is stored as its root operator alone,
+//! every node beneath it found or added as a memo group
+//! ([`over_child_groups`], Stage 0's node equality). Pass 1 and the
+//! Pass 2 rules store their candidates through [`operator_over_groups`].
+
+use std::collections::HashMap;
+use std::rc::Rc;
+
+use asap_types::ir::OperatorNode;
+use asap_types::stages::{Memo, MemoGroupId, OriginOfCandidateForMemoGroup};
+
+use crate::added_groups::groups_read_through_added_groups;
+use crate::find_or_add_group::over_child_groups;
+use crate::own_candidate_sizing::size_own_candidate_for;
+
+/// `root`, a candidate of a group `queries` reach, as one operator over
+/// groups: the exact guarantees below it removed
+/// ([`without_exact_guarantees_below`]), each child found or added with
+/// every node below it ([`over_child_groups`]: an added group is made
+/// by `origin`, sized for `accuracy_budget_parts`), and every added group it
+/// reads through sized for at least `accuracy_budget_parts` (a reader sized
+/// for more parts may find a group added for fewer). Returns the operator
+/// over the groups' nodes and those groups by child position.
+pub(crate) fn operator_over_groups(
+    memo: &mut Memo,
+    root: &Rc<OperatorNode>,
+    queries: &[usize],
+    origin: &OriginOfCandidateForMemoGroup,
+    accuracy_budget_parts: usize,
+) -> (Rc<OperatorNode>, Vec<MemoGroupId>) {
+    let root = without_exact_guarantees_below(root);
+    let (node, children) = over_child_groups(memo, &root, queries, origin, accuracy_budget_parts);
+    for group in groups_read_through_added_groups(memo, &children).added_groups {
+        size_own_candidate_for(memo, group, accuracy_budget_parts);
+    }
+    (node, children)
+}
+
+/// `root` with the exact guarantee removed from every node below it. Stage
+/// 0's nodes carry no guarantee, and composition gives each exact node it
+/// rebuilds the exact one ("exact over exact inputs"); that difference alone
+/// must not split a composed node from the equal Stage 0 group
+/// ([`find_or_add_group`](crate::find_or_add_group::find_or_add_group)
+/// compares guarantees). An approximate
+/// guarantee is kept, so builds that differ only in their failure
+/// probability or guarantee stay different groups. `root`'s own guarantee
+/// is the candidate's and is kept.
+pub(crate) fn without_exact_guarantees_below(root: &Rc<OperatorNode>) -> Rc<OperatorNode> {
+    fn without_exact(
+        node: &Rc<OperatorNode>,
+        done: &mut HashMap<*const OperatorNode, Rc<OperatorNode>>,
+    ) -> Rc<OperatorNode> {
+        if let Some(found) = done.get(&Rc::as_ptr(node)) {
+            return Rc::clone(found);
+        }
+        let operator = node
+            .operator
+            .map_children(|child| without_exact(child, done));
+        let exact = node.guarantee.as_ref().is_some_and(|g| g.is_exact());
+        let changed = exact
+            || node
+                .children()
+                .iter()
+                .zip(operator.children())
+                .any(|(a, b)| !Rc::ptr_eq(a, b));
+        let rebuilt = if changed {
+            Rc::new(OperatorNode {
+                operator,
+                guarantee: if exact { None } else { node.guarantee.clone() },
+                ..(**node).clone()
+            })
+        } else {
+            Rc::clone(node)
+        };
+        done.insert(Rc::as_ptr(node), Rc::clone(&rebuilt));
+        rebuilt
+    }
+    let mut done = HashMap::new();
+    let operator = root
+        .operator
+        .map_children(|child| without_exact(child, &mut done));
+    if root
+        .children()
+        .iter()
+        .zip(operator.children())
+        .all(|(a, b)| Rc::ptr_eq(a, b))
+    {
+        return Rc::clone(root);
+    }
+    Rc::new(OperatorNode {
+        operator,
+        ..(**root).clone()
+    })
+}