refactor: name sketch and population statistic requests

crates/asap-aware-mapping/src/accuracy/estimators/cardinality.rs

@@ -4,7 +4,7 @@ use super::*;
 pub(super) fn guarantee(
     algorithm: &SketchAlgorithm,
     params: &SketchParams,
-    query: &SketchQuery,
+    query: &SketchStatistic,
 ) -> Option<ResultGuarantee> {
     let (SketchParams::Kmv { k } | SketchParams::Theta { k }) = params else {
         return None;

crates/asap-aware-mapping/src/accuracy/estimators/cms.rs

@@ -4,7 +4,7 @@ use super::*;
 pub(super) fn guarantee(
     algorithm: &SketchAlgorithm,
     params: &SketchParams,
-    query: &SketchQuery,
+    query: &SketchStatistic,
 ) -> Option<ResultGuarantee> {
     let (SketchParams::Cms { width, depth } | SketchParams::CmsWithHeap { width, depth, .. }) =
         params
@@ -51,7 +51,7 @@ mod tests {
                     SketchKind::new(SketchAlgorithm::Cms, params),
                     GroupingStrategy::default(),
                 ),
-                &SketchQuery::Cardinality,
+                &SketchStatistic::Cardinality,
             )
             .unwrap();
         assert_eq!(g.metric, ErrorMetric::Frequency);
@@ -78,7 +78,7 @@ mod tests {
                     SketchKind::new(SketchAlgorithm::CmsWithHeap, cms_heap),
                     GroupingStrategy::default(),
                 ),
-                &SketchQuery::TopK { k: 10 },
+                &SketchStatistic::TopK { k: 10 },
             )
             .expect("heap sketch still provides per-key frequency intervals");
         assert_eq!(topk_frequency.metric, ErrorMetric::Frequency);

crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs

@@ -4,7 +4,7 @@ use super::*;
 pub(super) fn guarantee(
     algorithm: &SketchAlgorithm,
     params: &SketchParams,
-    query: &SketchQuery,
+    query: &SketchStatistic,
 ) -> Option<ResultGuarantee> {
     let (SketchParams::CountSketch { width, depth }
     | SketchParams::CountSketchWithHeap { width, depth, .. }) = params
@@ -66,7 +66,7 @@ mod tests {
                     SketchKind::new(SketchAlgorithm::CountSketch, count_sketch),
                     GroupingStrategy::default(),
                 ),
-                &SketchQuery::PointCount {
+                &SketchStatistic::PointCount {
                     key: asap_types::pre_asap::expr_ir::ColumnRef::SampleValue,
                     value: None,
                 },

crates/asap-aware-mapping/src/accuracy/estimators/ddsketch.rs

@@ -4,7 +4,7 @@ use super::*;
 pub(super) fn guarantee(
     algorithm: &SketchAlgorithm,
     params: &SketchParams,
-    query: &SketchQuery,
+    query: &SketchStatistic,
 ) -> Option<ResultGuarantee> {
     let SketchParams::DDSketch { alpha } = params else {
         return None;

crates/asap-aware-mapping/src/accuracy/estimators/hll.rs

@@ -9,7 +9,7 @@ use super::*;
 pub(super) fn generic_guarantee(
     algorithm: &SketchAlgorithm,
     params: &SketchParams,
-    query: &SketchQuery,
+    query: &SketchStatistic,
 ) -> Option<ResultGuarantee> {
     let SketchParams::Hll { precision } = params else {
         return None;
@@ -241,7 +241,7 @@ mod tests {
                     SketchKind::new(SketchAlgorithm::Hll, params),
                     GroupingStrategy::default(),
                 ),
-                &SketchQuery::Cardinality,
+                &SketchStatistic::Cardinality,
             )
             .unwrap();
         assert_eq!(g.metric, ErrorMetric::Cardinality);

crates/asap-aware-mapping/src/accuracy/estimators/kll.rs

@@ -4,7 +4,7 @@ use super::*;
 pub(super) fn guarantee(
     algorithm: &SketchAlgorithm,
     params: &SketchParams,
-    query: &SketchQuery,
+    query: &SketchStatistic,
 ) -> Option<ResultGuarantee> {
     let SketchParams::Kll { k } = params else {
         return None;
@@ -53,7 +53,7 @@ mod tests {
                     SketchKind::new(SketchAlgorithm::Kll, params),
                     GroupingStrategy::default(),
                 ),
-                &SketchQuery::Quantile { q: 0.99 },
+                &SketchStatistic::Quantile { q: 0.99 },
             )
             .unwrap();
         assert_eq!(g.metric, ErrorMetric::Rank);

crates/asap-aware-mapping/src/accuracy/estimators/mod.rs

@@ -14,7 +14,7 @@ pub mod univmon;
 pub(super) fn sketch_guarantee(
     algorithm: &SketchAlgorithm,
     params: &SketchParams,
-    query: &SketchQuery,
+    query: &SketchStatistic,
 ) -> Option<ResultGuarantee> {
     match params {
         SketchParams::Kll { .. } => kll::guarantee(algorithm, params, query),
@@ -36,7 +36,7 @@ pub(super) fn sketch_guarantee(
 fn bounded_guarantee(
     algorithm: &SketchAlgorithm,
     params: &SketchParams,
-    query: &SketchQuery,
+    query: &SketchStatistic,
     metric: ErrorMetric,
     bound: f64,
     delta: ProbabilityExpr,
@@ -57,7 +57,7 @@ fn bounded_guarantee(
 
 pub(super) fn local_guarantee(
     family: &FieldDataType,
-    query: &SketchQuery,
+    query: &SketchStatistic,
 ) -> Option<ResultGuarantee> {
     match family {
         FieldDataType::Plain(_) => Some(ResultGuarantee::exact("Plain value")),
@@ -196,9 +196,9 @@ impl AccuracyModel for EstimatorAccuracy<'_> {
     fn local_guarantee(
         &self,
         family: &FieldDataType,
-        query: &SketchQuery,
+        query: &SketchStatistic,
     ) -> Option<ResultGuarantee> {
-        if let (Some(_), FieldDataType::Sketch(kind, grouping), SketchQuery::Cardinality) =
+        if let (Some(_), FieldDataType::Sketch(kind, grouping), SketchStatistic::Cardinality) =
             (self.contract, family, query)
         {
             if let (SketchAlgorithm::Hll, SketchParams::Hll { precision }) =

crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs

@@ -1,8 +1,8 @@
 //! UnivMon currently certifies only its exact unit-update total readout.
 use super::*;
 
-pub(super) fn guarantee(query: &SketchQuery) -> Option<ResultGuarantee> {
-    matches!(query, SketchQuery::PointCount { value: None, .. })
+pub(super) fn guarantee(query: &SketchStatistic) -> Option<ResultGuarantee> {
+    matches!(query, SketchStatistic::PointCount { value: None, .. })
         .then(|| ResultGuarantee::exact("univmon_unit_update_total"))
 }
 

crates/asap-aware-mapping/src/accuracy/evidence.rs

@@ -119,7 +119,7 @@ pub trait AccuracyEvidenceProvider {
         &self,
         _op: &CompositionOperator,
         _family: &FieldDataType,
-        _query: Option<&SketchQuery>,
+        _query: Option<&SketchStatistic>,
     ) -> PropagationStats {
         PropagationStats::default()
     }
@@ -143,7 +143,7 @@ impl AccuracyEvidenceProvider for WorkloadAccuracyEvidence<'_> {
         &self,
         _op: &CompositionOperator,
         _family: &FieldDataType,
-        _query: Option<&SketchQuery>,
+        _query: Option<&SketchStatistic>,
     ) -> PropagationStats {
         PropagationStats {
             input_row_count: self.data.input_cardinality.value_at(self.now_ms).copied(),

crates/asap-aware-mapping/src/accuracy/mod.rs

@@ -22,7 +22,8 @@ pub use evidence::{
 
 use asap_types::post_asap::{
     AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, ExactOperation, FieldDataType,
-    GuaranteeSource, ProbabilityExpr, ResultGuarantee, SketchAlgorithm, SketchParams, SketchQuery,
+    GuaranteeSource, ProbabilityExpr, ResultGuarantee, SketchAlgorithm, SketchParams,
+    SketchStatistic,
 };
 use asap_types::types::AccuracyTarget;
 
@@ -48,7 +49,7 @@ pub trait AccuracyModel {
     fn local_guarantee(
         &self,
         family: &FieldDataType,
-        query: &SketchQuery,
+        query: &SketchStatistic,
     ) -> Option<ResultGuarantee>;
 
     /// Compose `inputs`' guarantees (in the parent's child order) with the
@@ -81,7 +82,7 @@ impl DefaultAccuracyModel {
     pub fn sketch_guarantee(
         algorithm: &SketchAlgorithm,
         params: &SketchParams,
-        query: &SketchQuery,
+        query: &SketchStatistic,
     ) -> Option<ResultGuarantee> {
         estimators::sketch_guarantee(algorithm, params, query)
     }
@@ -94,7 +95,7 @@ impl AccuracyModel for DefaultAccuracyModel {
     fn local_guarantee(
         &self,
         family: &FieldDataType,
-        query: &SketchQuery,
+        query: &SketchStatistic,
     ) -> Option<ResultGuarantee> {
         estimators::local_guarantee(family, query)
     }

crates/asap-aware-mapping/src/cost_model.rs

@@ -50,7 +50,7 @@ use std::rc::Rc;
 
 use asap_types::post_asap::{
     ExactOperation, FieldDataType, GroupingStrategy, HydraParams, ResultGuarantee, SketchAlgorithm,
-    SketchParams, SketchQuery, SummaryExpr, SummaryMaintenanceLifecycleGuarantee, SummaryNode,
+    SketchParams, SketchStatistic, SummaryExpr, SummaryMaintenanceLifecycleGuarantee, SummaryNode,
     SummaryWindowFramework,
 };
 use asap_types::pre_asap::agg_intent::AggIntent;
@@ -550,7 +550,7 @@ pub trait CostModel {
     /// [`realize_extension`](Self::realize_extension). Only ever called
     /// when `realize_extension` returned `Sketch` for the same
     /// `(ext_kind, payload)` — `replacement::readout` has no other way to build a
-    /// `SketchQuery` for a shape core doesn't know. A deployment that
+    /// `SketchStatistic` for a shape core doesn't know. A deployment that
     /// overrides `realize_extension` to return `Sketch` for some
     /// `ext_kind` MUST also override this for that same `ext_kind`, or
     /// this default panics loudly (rather than silently misinterpreting
@@ -560,7 +560,7 @@ pub trait CostModel {
         ext_kind: &str,
         _payload: &serde_json::Value,
         _col: &ColumnRef,
-    ) -> SketchQuery {
+    ) -> SketchStatistic {
         unimplemented!(
             "CostModel::realize_extension returned Sketch for ext_kind={ext_kind:?} but \
              readout_extension wasn't overridden to match"

crates/asap-aware-mapping/src/grouping.rs

@@ -673,7 +673,7 @@ mod tests {
             &self,
             _op: &CompositionOperator,
             _family: &FieldDataType,
-            _query: Option<&asap_types::post_asap::SketchQuery>,
+            _query: Option<&asap_types::post_asap::SketchStatistic>,
         ) -> PropagationStats {
             PropagationStats {
                 hydra_shared_grid_collision_bound: Some(0.0),
@@ -717,7 +717,7 @@ mod tests {
                 &self,
                 _op: &CompositionOperator,
                 _family: &FieldDataType,
-                _query: Option<&asap_types::post_asap::SketchQuery>,
+                _query: Option<&asap_types::post_asap::SketchStatistic>,
             ) -> PropagationStats {
                 PropagationStats {
                     hydra_shared_grid_failure_probability: Some(1.5),
@@ -763,7 +763,7 @@ mod tests {
                 &self,
                 _op: &CompositionOperator,
                 _family: &FieldDataType,
-                _query: Option<&asap_types::post_asap::SketchQuery>,
+                _query: Option<&asap_types::post_asap::SketchStatistic>,
             ) -> PropagationStats {
                 PropagationStats {
                     hydra_shared_grid_collision_bound: Some(0.1),

crates/asap-aware-mapping/src/maintained_population.rs

@@ -23,7 +23,9 @@ fn strip_projection(mut root: &QueryExpr) -> &QueryExpr {
     root
 }
 
-fn recognize(root: &QueryExpr) -> Option<(MaintainedPopulation, PopulationReadout, Rc<QueryExpr>)> {
+fn recognize(
+    root: &QueryExpr,
+) -> Option<(MaintainedPopulation, PopulationStatistic, Rc<QueryExpr>)> {
     let root = strip_projection(root);
     let (source, grouping, readout, value_column) = match root {
         QueryExpr::Aggregate {
@@ -42,12 +44,12 @@ fn recognize(root: &QueryExpr) -> Option<(MaintainedPopulation, PopulationReadou
             }
             let (col, readout) = match intent {
                 AggIntent::Quantile { q, col, .. } if q.is_finite() => {
-                    (*col, PopulationReadout::Quantile { q: *q })
+                    (*col, PopulationStatistic::Quantile { q: *q })
                 }
-                AggIntent::TopK { k, .. } => (None, PopulationReadout::TopK { k: *k }),
-                AggIntent::Sum { col } => (*col, PopulationReadout::Sum),
-                AggIntent::Count { .. } => (None, PopulationReadout::Count),
-                AggIntent::Avg { col } => (*col, PopulationReadout::Average),
+                AggIntent::TopK { k, .. } => (None, PopulationStatistic::TopK { k: *k }),
+                AggIntent::Sum { col } => (*col, PopulationStatistic::Sum),
+                AggIntent::Count { .. } => (None, PopulationStatistic::Count),
+                AggIntent::Avg { col } => (*col, PopulationStatistic::Average),
                 _ => return None,
             };
             let schema = child.output_schema().ok()?;
@@ -81,7 +83,7 @@ fn recognize(root: &QueryExpr) -> Option<(MaintainedPopulation, PopulationReadou
             (
                 child,
                 partition_by,
-                PopulationReadout::TopK { k: *n },
+                PopulationStatistic::TopK { k: *n },
                 Some(*col),
             )
         }
@@ -237,11 +239,13 @@ impl MaintainedPopulationStrategy {
             if let Some((p, r, _)) = recognize(other) {
                 if p == identity {
                     match r {
-                        PopulationReadout::Quantile { .. } => population.quantiles = true,
-                        PopulationReadout::TopK { k } => population.max_k = population.max_k.max(k),
-                        PopulationReadout::Sum
-                        | PopulationReadout::Count
-                        | PopulationReadout::Average => {}
+                        PopulationStatistic::Quantile { .. } => population.quantiles = true,
+                        PopulationStatistic::TopK { k } => {
+                            population.max_k = population.max_k.max(k)
+                        }
+                        PopulationStatistic::Sum
+                        | PopulationStatistic::Count
+                        | PopulationStatistic::Average => {}
                     }
                 }
             }
@@ -469,7 +473,7 @@ mod tests {
             unreachable!()
         };
         *operation = ValueOperation::ReadPopulation {
-            readout: PopulationReadout::TopK { k: 6 },
+            readout: PopulationStatistic::TopK { k: 6 },
         };
         assert!(compile_post_asap_dag(&Rc::new(bad.clone())).is_err());
         let SummaryExpr::ValueOperation {
@@ -479,7 +483,7 @@ mod tests {
             unreachable!()
         };
         *operation = ValueOperation::ReadPopulation {
-            readout: PopulationReadout::TopK { k: 5 },
+            readout: PopulationStatistic::TopK { k: 5 },
         };
         let producer = Rc::make_mut(child);
         let SummaryExpr::ValueOperation {