fix(types,executor): one family-capability table decides which statistic a family answers

crates/devtools/tests/stage_pipeline.rs

@@ -1510,11 +1510,9 @@ fn display_cap_shows_the_selected_then_valid_cheapest_then_each_cause() {
     assert!(costs.len() >= 2, "{costs:?}");
 }
 
-/// With no supported summaries declared, `stage_pipeline` plans with the
-/// executor's families, as `plan_installable` does: a count's CountSketch,
-/// which the Planner's executor does not run, is pruned in Pass 1 instead of
-/// offered (before, `Q1 CountSketch` was a candidate), and the document
-/// shows the families used. A declared list is kept as declared.
+/// With no supported summaries declared, `stage_pipeline` uses the executor's
+/// families, as `plan_installable` does. CountSketch cannot answer a total, so
+/// it is neither generated for count nor reported as deployment-pruned.
 #[test]
 fn undeclared_supported_summaries_default_to_the_executors_families() {
     let document = document_of(&[
@@ -1540,8 +1538,8 @@ fn undeclared_supported_summaries_default_to_the_executors_families() {
     }
     assert!(labels.iter().all(|label| !label.contains("CountSketch")));
     assert_eq!(
-        document["stage1_logical_asap_optimization"]["pass1_pruned"][0]["pruned"],
-        serde_json::json!([{ "candidate": "CountSketch", "reason": "UnsupportedByDeployment" }])
+        document["stage1_logical_asap_optimization"]["pass1_pruned"],
+        serde_json::json!([])
     );
     assert_eq!(
         document["deployment_capabilities"]["supported_summaries"],

crates/executor/src/capability.rs

@@ -12,8 +12,8 @@ use crate::summary_kernels::hll_sketch::HLL_PRECISIONS;
 use crate::summary_kernels::weighted_frequency::{self, FrequencyAlgorithm};
 use crate::Error;
 use planner_types::ir::schema::{
-    ExactKind, ExactParams, FieldDataType as SummaryFamilyType, GroupingStrategy, SketchAlgorithm,
-    SketchParams, SketchStatistic, SummaryUpdate,
+    ExactKind, ExactParams, FamilyCapability, FieldDataType as SummaryFamilyType, GroupingStrategy,
+    SketchAlgorithm, SketchParams, SketchStatistic, StatisticKind, SummaryUpdate,
 };
 
 /// Check the same contract used by `create_planner_accumulator` before a plan
@@ -341,67 +341,90 @@ pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error> {
     .map_err(Error::Invalid)
 }
 
-/// A sketch evaluation is native only for the families Planner can read directly.
+/// The families the executor builds per instance and reads, in
+/// `SketchAlgorithm` order. A family it builds but reads nothing from
+/// (plain Count Sketch) and one it does not build (Theta) are absent.
+pub(crate) const EVALUATED_FAMILIES: [SketchAlgorithm; 8] = [
+    SketchAlgorithm::UnivMon,
+    SketchAlgorithm::Kll,
+    SketchAlgorithm::Cms,
+    SketchAlgorithm::Hll,
+    SketchAlgorithm::DDSketch,
+    SketchAlgorithm::CmsWithHeap,
+    SketchAlgorithm::Kmv,
+    SketchAlgorithm::CountSketchWithHeap,
+];
+
+/// The readouts the family-capability table lists for an evaluated family
+/// that the executor does not evaluate yet: a per-instance Count-Min's item
+/// frequency (the executor answers an item's frequency only from a Hydra
+/// grid, which Stage 2 does not produce yet; `SupportedSummary` gains a
+/// grouping dimension when it does).
+pub(crate) const UNEVALUATED_READOUTS: [(SketchAlgorithm, StatisticKind); 1] =
+    [(SketchAlgorithm::Cms, StatisticKind::ItemFrequency)];
+
+/// Whether the executor reads `readout` from a state of `family` whose
+/// family-capability table row answers it.
+fn evaluates(family: &SummaryFamilyType, readout: StatisticKind) -> bool {
+    use planner_types::ir::schema::HydraKind;
+    match family {
+        SummaryFamilyType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance) => {
+            let algorithm = kind.algorithm();
+            EVALUATED_FAMILIES.contains(algorithm)
+                && !UNEVALUATED_READOUTS.contains(&(algorithm.clone(), readout))
+        }
+        SummaryFamilyType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind, .. }) => {
+            matches!(kind, HydraKind::HydraCms | HydraKind::HydraKll)
+        }
+        _ => false,
+    }
+}
+
+/// A sketch evaluation is native when the family-capability table says the
+/// family answers the statistic and the executor evaluates that readout.
+/// Top-k is a keyed evaluation (`Operator::keyed_evaluation`), not a sketch
+/// evaluation.
 pub fn validate_sketch_evaluation(
     family: &SummaryFamilyType,
     query: &SketchStatistic,
 ) -> Result<(), Error> {
     validate_native_family(family)?;
-    use planner_types::ir::schema::SketchAlgorithm as A;
-    // A point count without an item value reads the total count.
-    let bare_count = matches!(query, SketchStatistic::PointCount { value: None, .. });
-    let supported = match family {
-        SummaryFamilyType::Sketch(kind, _) => match (kind.algorithm(), query) {
-            (A::Kll, SketchStatistic::Quantile { q })
-            | (A::DDSketch, SketchStatistic::Quantile { q }) => {
-                if !(0.0..=1.0).contains(q) {
-                    return Err(Error::Invalid(
-                        "quantile evaluation requires quantile in [0,1]".into(),
-                    ));
-                }
-                true
-            }
-            (A::DDSketch, _) => bare_count,
-            (A::Hll, SketchStatistic::Cardinality) => true,
-            (A::Hll, _) => bare_count,
-            // HydraCms answers a group's item frequency. A group's total
-            // needs a group-mass query asap_sketchlib's Hydra lacks.
-            (A::Cms, SketchStatistic::PointCount { value, .. })
-                if matches!(family, SummaryFamilyType::Sketch(_, grouping)
-                    if grouping != &GroupingStrategy::PerSubpopulationInstance) =>
-            {
-                if value.is_none() {
-                    return Err(Error::Invalid(
-                        "HydraCms does not answer a group total: asap_sketchlib's Hydra has no \
-                         group-mass query; only an item's point count within a group is answered"
-                            .into(),
-                    ));
+    let readout = query.kind();
+    // HydraCms answers a group's item frequency. A group's total needs a
+    // group-mass query asap_sketchlib's Hydra lacks.
+    if readout == StatisticKind::Total
+        && matches!(
+            family,
+            SummaryFamilyType::Sketch(
+                _,
+                GroupingStrategy::SharedMultiSubpopulation {
+                    kind: planner_types::ir::schema::HydraKind::HydraCms,
+                    ..
                 }
-                true
-            }
-            (A::Kmv, SketchStatistic::Cardinality) => true,
-            (A::Kmv, _) => bare_count,
-            // One frequency state answers its total, distinct count, L2 norm
-            // and entropy.
-            (
-                A::UnivMon,
-                SketchStatistic::Cardinality
-                | SketchStatistic::FrequencyL2
-                | SketchStatistic::FrequencyEntropy,
-            ) => true,
-            (A::UnivMon, _) => bare_count,
-            // Only count intents read a Count-Min bare count, and their
-            // updates have unit weight; the evaluation is typed Int64 on that basis.
-            (A::Cms, _) => bare_count,
-            _ => false,
-        },
-        _ => false,
-    };
-    if !supported {
+            )
+        )
+    {
+        return Err(Error::Invalid(
+            "HydraCms does not answer a group total: asap_sketchlib's Hydra has no group-mass \
+             query; only an item's point count within a group is answered"
+                .into(),
+        ));
+    }
+    if readout == StatisticKind::TopK
+        || !FamilyCapability::of(family).answers(query)
+        || !evaluates(family, readout)
+    {
         return Err(Error::Invalid(
             "evaluation is not implemented for this summary family".into(),
         ));
     }
+    if let SketchStatistic::Quantile { q } = query {
+        if !(0.0..=1.0).contains(q) {
+            return Err(Error::Invalid(
+                "quantile evaluation requires quantile in [0,1]".into(),
+            ));
+        }
+    }
     Ok(())
 }
 

crates/executor/src/summary_kernels/dd_sketch.rs

@@ -42,7 +42,7 @@ impl AggregateCore for DDSketchAccumulator {
         self.inner.get_count() == 0
     }
 
-    /// Quantiles, and the total sample count as a bare `PointCount`.
+    /// Quantiles.
     fn estimate(&self, query: &SketchStatistic) -> Result<f64, KernelError> {
         match query {
             // PromQL `quantile_over_time` and SQL `percentile_cont`
@@ -55,7 +55,6 @@ impl AggregateCore for DDSketchAccumulator {
                 .inner
                 .quantile_interpolated(*q)?
                 .ok_or_else(|| "DDSketch quantile of an empty population".into()),
-            SketchStatistic::PointCount { value: None, .. } => Ok(self.inner.get_count() as f64),
             other => Err(format!("DDSketch does not answer {other:?}").into()),
         }
     }
@@ -73,9 +72,9 @@ mod tests {
         }
     }
 
-    // A bare point count reads the total sample count, and merge adds counts.
+    // A merged state's quantile is the union's.
     #[test]
-    fn count_and_quantile_survive_merge() {
+    fn quantile_survives_merge() {
         let (mut a, mut b) = (
             DDSketchAccumulator::new(0.01),
             DDSketchAccumulator::new(0.01),
@@ -85,19 +84,20 @@ mod tests {
             b.inner.add(&f64::from(v + 50));
         }
         let merged = a.merge_with(&b).unwrap();
-        assert_eq!(merged.estimate(&bare_count()).unwrap(), 100.0);
         let median = merged
             .estimate(&SketchStatistic::Quantile { q: 0.5 })
             .unwrap();
         assert!((median - 50.0).abs() <= 1.0, "{median}");
     }
 
-    // An empty DDSketch has no quantile, and unsupported queries are errors.
+    // An empty DDSketch has no quantile, and unsupported queries (its
+    // total among them) are errors.
     #[test]
     fn empty_quantile_and_unsupported_queries_fail() {
         let dd = DDSketchAccumulator::new(0.01);
         assert!(dd.estimate(&SketchStatistic::Quantile { q: 0.5 }).is_err());
         assert!(dd.estimate(&SketchStatistic::Cardinality).is_err());
+        assert!(dd.estimate(&bare_count()).is_err());
     }
 
     // The quantile at q interpolates linearly between the order statistics

crates/executor/src/summary_kernels/hll_sketch.rs

@@ -155,13 +155,10 @@ impl AggregateCore for HllSketchAccumulator {
         Ok(Box::new(merged))
     }
 
-    /// Distinct count, rounded down to an integer. A bare `PointCount` over
-    /// an HLL also reads the distinct count.
+    /// Distinct count, rounded down to an integer.
     fn estimate(&self, query: &SketchStatistic) -> Result<f64, KernelError> {
         match query {
-            SketchStatistic::Cardinality | SketchStatistic::PointCount { value: None, .. } => {
-                Ok(self.inner.estimate() as f64)
-            }
+            SketchStatistic::Cardinality => Ok(self.inner.estimate() as f64),
             other => Err(format!("HLL does not answer {other:?}").into()),
         }
     }

crates/executor/src/summary_kernels/kmv.rs

@@ -62,12 +62,10 @@ impl AggregateCore for KmvAccumulator {
         Ok(Box::new(merged))
     }
 
-    /// Distinct count. A bare `PointCount` over a KMV also reads the distinct count.
+    /// Distinct count.
     fn estimate(&self, query: &SketchStatistic) -> Result<f64, KernelError> {
         match query {
-            SketchStatistic::Cardinality | SketchStatistic::PointCount { value: None, .. } => {
-                Ok(self.inner.estimate())
-            }
+            SketchStatistic::Cardinality => Ok(self.inner.estimate()),
             other => Err(format!("KMV does not answer {other:?}").into()),
         }
     }
@@ -150,14 +148,6 @@ mod tests {
     fn counts_fewer_than_k_distinct_items_exactly() {
         let kmv = build(1024, (0..500).chain(0..500));
         assert_eq!(cardinality(&kmv), 500.0);
-        assert_eq!(
-            kmv.estimate(&SketchStatistic::PointCount {
-                key: planner_types::ir::scalar::ColumnRef::SampleValue,
-                value: None,
-            })
-            .unwrap(),
-            500.0
-        );
     }
 
     // Above k the estimate is within ten times KMV's relative standard

crates/executor/src/summary_kernels/traits.rs

@@ -68,10 +68,9 @@ mod tests {
             .downcast_mut::<DDSketchAccumulator>()
             .unwrap();
         dd.inner.add(&3.0);
-        let count = SketchStatistic::PointCount {
-            key: planner_types::ir::scalar::ColumnRef::SampleValue,
-            value: None,
-        };
-        assert_eq!(state.estimate(&count).unwrap(), 1.0);
+        let median = state
+            .estimate(&SketchStatistic::Quantile { q: 0.5 })
+            .unwrap();
+        assert!((median - 3.0).abs() <= 0.03, "{median}");
     }
 }