feat(summary_executor): read out named-key PointCount (count(metric{item=x}))

data_plane/Cargo.toml

@@ -24,12 +24,17 @@ control_plane = { path = "../control_plane" }
 # `crates/asap_types/Cargo.toml`'s existing pin comment: two revs of the
 # same git dependency in one workspace resolve to two distinct Rust
 # types that won't unify.
-asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "d4c175633f6ce46b801ca2115c00fa757d5b6240" }
+#
+# Bumped to 64df20d (main tip, ASAPController#162, merged -- matches
+# control_plane's own pin) to pick up
+# `SketchQuery::PointCount.value: Option<String>`, needed for the
+# named-key PointCount readout.
+asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
 # `asap_sketch::L4Node`'s own fields (`SummaryExpr::Logical(Box<QueryExpr>)`,
 # `SummaryAgg { col: ColumnRef, by: Vec<ColumnId>, .. }`) are `asap-ir`
 # types, not re-exported by `asap-sketch` -- `find_candidates` needs to
 # walk/match them directly. Same pin-must-match rule as `asap-sketch` above.
-asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "d4c175633f6ce46b801ca2115c00fa757d5b6240" }
+asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" }
 
 # Shared external (workspace)
 serde.workspace = true

data_plane/src/query_engines/asap_query_engine/summary_executor.rs

@@ -7,13 +7,18 @@
 //! ## Scope
 //!
 //! Covers **quantile/cardinality queries** (DDSketch/Kll/Hll) **and the
-//! Frequency family's bare total** (CMS/CountSketch/CMS-with-heap/
-//! CountSketch-with-heap, `count`/`sum` with no specific item key), both
-//! cumulative (instant) and per-window (matrix/range). All modes do real
-//! cross-sid merging via `delta_apply::SummaryState`: reconstruct each
-//! candidate sid's own state over the range (or per window), then merge
-//! same-window/same-range states *across* sids before reading out one
-//! answer per group (or per group per window).
+//! Frequency family** — both the bare total (`count`/`sum` with no
+//! specific item key) and a per-item point lookup (`count(cms_metric
+//! {item="x"})`, `SketchQuery::PointCount{key: Named(_), value: Some(_)}`
+//! — `value` is where the filter's actual value lives; see
+//! `asap_sketch::SketchQuery::PointCount`'s doc for why `readout` can't
+//! resolve it itself from a `Filter` predicate) — for CMS/CountSketch/
+//! CMS-with-heap/CountSketch-with-heap, both cumulative (instant) and
+//! per-window (matrix/range). All modes do real cross-sid merging via
+//! `delta_apply::SummaryState`: reconstruct each candidate sid's own
+//! state over the range (or per window), then merge same-window/
+//! same-range states *across* sids before reading out one answer per
+//! group (or per group per window).
 //!
 //! `Self::Value` (`SummaryValue`) carries two shapes: `Points` (one
 //! scalar per timestamp — everything but `TopK`) and `TopK` (one ranked
@@ -29,15 +34,12 @@
 //! - `SketchQuery::TopK` against a heap-less family (`Dd`/`Hll`/`Kll`/
 //!   `Cms`/`CountSketch`) — a family limitation (no item universe to
 //!   rank), not an unimplemented-query limitation; see `topk_ranked`.
-//! - `SketchQuery::PointCount` with a *named* item key (a point lookup
-//!   for one specific item, e.g. `count(cms_metric{item="x"})`). The
-//!   *value* to look up isn't carried by `SketchQuery` or available in
-//!   `readout`'s signature — `PointCount{key: ColumnRef}` names which
-//!   *column* is being queried, not the value to filter for, which would
-//!   come from a `Filter` predicate elsewhere in the tree. Resolving
-//!   that is a separate problem from this trait's scope.
-//!   `PointCount{key: ColumnRef::SampleValue}` (no specific item — the
-//!   bare bucket total) is covered.
+//! - `SketchQuery::PointCount` against a heap-less-*and*-quantile/
+//!   cardinality family (`Dd`/`Hll`/`Kll` have no item universe at all)
+//!   — a family limitation, not an unimplemented-query limitation.
+//! - A `PointCount` whose `key`/`value` combination isn't one of the two
+//!   expected shapes (`SampleValue` + `None`, or `Named`/`Qualified` +
+//!   `Some(_)`) — reported rather than silently guessed at.
 //! - `ExactAgg` intents (`Sum`/`Rate`/`Increase`/`MinMax`/exact `Count`).
 //!   These don't reach `readout` at all — `asap_plan::bind` never wraps
 //!   an `ExactAccumulator` implementation in a `SummaryEstimate`
@@ -368,15 +370,27 @@ fn sketch_query_value(rs: &SummaryState, query: &SketchQuery) -> Result<f64, Sum
     match query {
         SketchQuery::Quantile { q } => Ok(rs.quantile(*q)),
         SketchQuery::Cardinality => Ok(rs.cardinality()),
-        // `key: ColumnRef::SampleValue` means "no specific item" -- the
-        // bare bucket total. Any other column names an item to look up
-        // by VALUE, which isn't carried by `SketchQuery` -- see the
-        // module doc.
+        // `key: ColumnRef::SampleValue, value: None` means "no specific
+        // item" -- the bare bucket total. `key: Named(_), value: Some(v)`
+        // is a per-item point lookup (e.g. `count(cms_metric{item="x"})`)
+        // -- `value` is where the filter's actual value lives (see
+        // `asap_sketch::SketchQuery::PointCount`'s doc for why `readout`
+        // can't resolve it itself). Any other combination (e.g. a `Named`
+        // key with no value, or `SampleValue` with a value) is a shape
+        // this executor doesn't expect to see and reports rather than
+        // silently misreading.
         SketchQuery::PointCount {
             key: ColumnRef::SampleValue,
+            value: None,
         } => Ok(rs.total()),
+        SketchQuery::PointCount {
+            key: ColumnRef::Named(_) | ColumnRef::Qualified { .. },
+            value: Some(v),
+        } => rs.estimate(v).ok_or(SummaryExecutorError::Unsupported(
+            "PointCount by key requires a Frequency-family sketch (Cms/CountSketch/..WithHeap)",
+        )),
         SketchQuery::PointCount { .. } => Err(SummaryExecutorError::Unsupported(
-            "PointCount for a named item key needs a filter value this trait doesn't carry",
+            "unrecognized PointCount shape (key/value combination not expected)",
         )),
         // Both readout callers branch on `TopK` before ever calling this
         // function (see `readout_cumulative`/`readout_per_window`), so
@@ -774,6 +788,17 @@ mod tests {
         sk.to_msgpack().expect("encode CountMinSketch msgpack")
     }
 
+    /// Encode a CMS msgpack frame with one `update` of `weight` for a
+    /// single named `key` -- unlike `encode_cms_with_total`'s synthetic
+    /// "k", this lets a test control which key a `PointCount` query looks
+    /// up.
+    fn encode_cms_with_item(rows: usize, cols: usize, key: &str, weight: f64) -> Vec<u8> {
+        use asap_sketchlib::{CountMinSketch, MessagePackCodec};
+        let mut sk = CountMinSketch::new(rows, cols);
+        sk.update(key, weight);
+        sk.to_msgpack().expect("encode CountMinSketch msgpack")
+    }
+
     fn cms_agg_node(child: Rc<L4Node>) -> Rc<L4Node> {
         Rc::new(L4Node {
             expr: SummaryExpr::SummaryAgg {
@@ -1090,6 +1115,7 @@ mod tests {
             cms_agg_node(child),
             SketchQuery::PointCount {
                 key: ColumnRef::SampleValue,
+                value: None,
             },
         );
 
@@ -1139,6 +1165,7 @@ mod tests {
             cms_agg_node(child),
             SketchQuery::PointCount {
                 key: ColumnRef::SampleValue,
+                value: None,
             },
         );
 
@@ -1157,6 +1184,97 @@ mod tests {
         );
     }
 
+    #[test]
+    fn single_cms_sid_named_key_point_estimate() {
+        let idx = SketchStore::new();
+        let sid = 1u64;
+        idx.register(cms_meta(sid, "requests_by_route"));
+        idx.append_sample(
+            sid,
+            BTreeMap::new(),
+            (T0, T0 + 1000),
+            SketchSampleState {
+                bytes: encode_cms_with_item(4, 256, "checkout", 17.0),
+                encoding: crate::storage_engines::sketch_db::index::SketchEncoding::MsgpackFull,
+            },
+        );
+
+        let child = scan_node("requests_by_route", None);
+        let tree = estimate_node(
+            cms_agg_node(child),
+            SketchQuery::PointCount {
+                key: ColumnRef::Named("item".to_string()),
+                value: Some("checkout".to_string()),
+            },
+        );
+
+        let exec = ctx(&idx);
+        let ExecOutcome::Value(v) = execute(&tree, &exec).expect("execute should succeed") else {
+            panic!("expected a value");
+        };
+        let (_group, value) = &v[0];
+        let SummaryValue::Points(samples) = value else {
+            panic!("expected Points, got {value:?}");
+        };
+        let (_ts, estimate) = samples[0];
+        assert_eq!(
+            estimate, 17.0,
+            "named-key point estimate must equal that key's own insertions"
+        );
+    }
+
+    #[test]
+    fn two_cms_sids_same_group_named_key_estimate_merges_cross_sid() {
+        // Cross-sid merge for a NAMED-KEY point lookup: the same key's
+        // weight contributed by two different sids must ADD (matrix merge
+        // then keyed estimate), not just report one sid's contribution --
+        // the point-lookup analog of `two_cms_sids_same_group_totals_actually_merge`.
+        let idx = SketchStore::new();
+        idx.register(cms_meta(1, "requests_by_route"));
+        idx.register(cms_meta(2, "requests_by_route"));
+        idx.append_sample(
+            1,
+            BTreeMap::new(),
+            (T0, T0 + 1000),
+            SketchSampleState {
+                bytes: encode_cms_with_item(4, 256, "checkout", 30.0),
+                encoding: crate::storage_engines::sketch_db::index::SketchEncoding::MsgpackFull,
+            },
+        );
+        idx.append_sample(
+            2,
+            BTreeMap::new(),
+            (T0, T0 + 1000),
+            SketchSampleState {
+                bytes: encode_cms_with_item(4, 256, "checkout", 12.0),
+                encoding: crate::storage_engines::sketch_db::index::SketchEncoding::MsgpackFull,
+            },
+        );
+
+        let child = scan_node("requests_by_route", None);
+        let tree = estimate_node(
+            cms_agg_node(child),
+            SketchQuery::PointCount {
+                key: ColumnRef::Named("item".to_string()),
+                value: Some("checkout".to_string()),
+            },
+        );
+
+        let exec = ctx(&idx);
+        let ExecOutcome::Value(v) = execute(&tree, &exec).expect("execute should succeed") else {
+            panic!("expected a value");
+        };
+        let (_group, value) = &v[0];
+        let SummaryValue::Points(samples) = value else {
+            panic!("expected Points, got {value:?}");
+        };
+        let (_ts, estimate) = samples[0];
+        assert_eq!(
+            estimate, 42.0,
+            "merged named-key estimate must be the SUM of both sids (30 + 12)"
+        );
+    }
+
     #[test]
     fn topk_query_against_non_heap_sketch_is_unsupported() {
         // A heap-less family (Cms/CountSketch/Dd/Hll/Kll) carries no item

data_plane/src/storage_engines/sketch_db/query/delta_apply.rs

@@ -431,6 +431,21 @@ impl SummaryState {
             .unwrap_or(0.0)
     }
 
+    /// Per-key point estimate — `count(metric{item="x"})`-shaped queries.
+    /// Unlike [`Self::topk_items`], no heap is needed: all four Frequency
+    /// variants (heap-bearing or not) already carry a keyed `estimate`
+    /// over their matrix. `None` for the quantile/cardinality states,
+    /// which have no item universe at all.
+    pub fn estimate(&self, key: &str) -> Option<f64> {
+        match self {
+            SummaryState::Cms(c) => Some(c.estimate(key)),
+            SummaryState::CountSketch(c) => Some(c.estimate(key)),
+            SummaryState::CmsWithHeap(h) => Some(h.estimate(key)),
+            SummaryState::CountSketchWithHeap(h) => Some(h.estimate(key)),
+            _ => None,
+        }
+    }
+
     /// Top-k `(key, value)` pairs from the heap, descending by value.
     /// `None` for anything other than a heap-bearing state — the
     /// heap-less Frequency states (`Cms`/`CountSketch`) carry no item