feat(physical-operators): native UnivMon state over typed identities

UnivMon had a kernel but no native DAG state: a build was rejected
("summary family has no native DAG state implementation"), a summary
build accepted only Float64 samples, and Pass 1's UnivMon update (each
value of the item column one unit occurrence) did not bind, so no
UnivMon candidate of #509 Example 2 could execute.

- UnivMon is a native family; its states validate against their
  parameters, and one state answers its total, distinct count, L2 norm
  and entropy.
- A UnivMon build takes Bool, Int64 and Utf8 identities as type-tagged
  key bytes (Int64 above 2^53 stays distinct) besides Float64 samples;
  NULL is skipped. Retained heap key bytes count toward state memory.
- A unit-frequency update over one item column binds, for UnivMon and
  HLL, as a value-frequency build over that column: grouped (SQL
  src_ip) and per series (PromQL sample values).

Ports the UnivMon part of #597 (from #557).

Author: zzylol

crates/asap-physical-operators/src/capability.rs

@@ -112,7 +112,7 @@ pub fn validate_summary_kernel(
         }
         _ => return Err(format!("unsupported summary kernel {family:?}")),
     };
-    if keyed != input.item.is_some() && !is_unit_sample_frequency(input) {
+    if keyed != input.item.is_some() && unit_frequency_column(input).is_none() {
         return Err("Planner item expression does not match kernel layout".into());
     }
     Ok(())
@@ -217,6 +217,25 @@ fn valid_matrix(width: u32, depth: u32) -> bool {
             .is_some()
 }
 
+/// The one item column of a unit-count frequency update (each value is one
+/// occurrence), the update an unkeyed frequency family (HLL, UnivMon) builds
+/// from that column's values.
+pub(crate) fn unit_frequency_column(
+    update: &planner_types::post_asap::SummaryUpdate,
+) -> Option<&planner_types::pre_asap::ColumnRef> {
+    use planner_types::post_asap::{NonNegativeWeightProof, SummaryInputExpr, WeightDomain};
+    match (&update.item, &update.weight, &update.weight_domain) {
+        (
+            Some(SummaryInputExpr::Column(column)),
+            SummaryInputExpr::Constant(weight),
+            WeightDomain::NonNegative {
+                proof: NonNegativeWeightProof::UnitCount,
+            },
+        ) if *weight == 1.0 => Some(column),
+        _ => None,
+    }
+}
+
 pub(crate) fn is_unit_sample_frequency(update: &planner_types::post_asap::SummaryUpdate) -> bool {
     use planner_types::post_asap::{NonNegativeWeightProof, SummaryInputExpr, WeightDomain};
     matches!(
@@ -266,7 +285,10 @@ pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error> {
     match family {
         SummaryFamilyType::ExactAggregate(..) => {}
         SummaryFamilyType::Sketch(kind, _)
-            if matches!(kind.algorithm(), A::Kll | A::DDSketch | A::Hll | A::Kmv) => {}
+            if matches!(
+                kind.algorithm(),
+                A::Kll | A::DDSketch | A::Hll | A::Kmv | A::UnivMon
+            ) => {}
         _ => {
             return Err(Error::Invalid(
                 "summary family has no native DAG state implementation".into(),
@@ -323,6 +345,15 @@ pub fn validate_sketch_evaluation(
             }
             (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,

crates/asap-physical-operators/src/operators/summary/mod.rs

@@ -167,8 +167,14 @@ impl Operator {
     ) -> Result<Self, Error> {
         crate::values::validate_family(&family)?;
         validate_groups(&input, &groups)?;
-        if plain(&input, value)?.0 != &DataType::Float64 {
-            return Err(invalid("summary numeric update requires Float64"));
+        let dtype = plain(&input, value)?.0;
+        let typed_frequency = matches!(&family, SummaryFamilyType::Sketch(kind, _)
+            if kind.algorithm() == &planner_types::post_asap::SketchAlgorithm::UnivMon);
+        if *dtype != DataType::Float64
+            && !(typed_frequency
+                && matches!(dtype, DataType::Utf8 | DataType::Int64 | DataType::Bool))
+        {
+            return Err(invalid("summary update type is unsupported by its family"));
         }
         if let Some(time) = time {
             if plain(&input, time)? != (&DataType::Timestamp, false) {
@@ -532,11 +538,10 @@ async fn build_summary(
                 states.get_mut(&key).expect("inserted group");
             // SQL aggregates ignore NULL samples while retaining the group.
             // A missing counter sample also contributes no observation.
-            let value = match row[value] {
-                Value::Float64(value) => value,
-                Value::Null => continue,
-                _ => return Err(invalid("summary update type")),
-            };
+            let value = &row[value];
+            if matches!(value, Value::Null) {
+                continue;
+            }
             let timestamp = if let Some(time) = time {
                 let Value::Timestamp(time) = row[time] else {
                     return Err(invalid("summary time type"));
@@ -552,9 +557,8 @@ async fn build_summary(
                 ));
             }
             updater
-                .validate_single_input(value)
+                .update_value(value, timestamp)
                 .map_err(Error::Operator)?;
-            updater.update_single(value, timestamp);
             *previous = Some(timestamp);
             memory.resize(updater.memory_usage_bytes() + *overhead)?;
         }

crates/asap-physical-operators/src/physical_planner/mod.rs

@@ -427,17 +427,25 @@ fn compile_internal(
                 let Some(NonASAPOp::Scan { schema, .. }) = source.non_asap() else {
                     return Err(invalid("per-entity summary requires a resolved source"));
                 };
-                if !schema.closed || update.item.is_some() {
+                // An unkeyed frequency family (UnivMon) counts each value of
+                // its one item column once, as a value-frequency build.
+                let frequency_item = crate::capability::unit_frequency_column(update)
+                    .filter(|_| unkeyed_frequency_family(family));
+                if !schema.closed || (update.item.is_some() && frequency_item.is_none()) {
                     return Err(invalid(
                         "per-entity summary requires complete source identity",
                     ));
                 }
                 crate::capability::validate_summary_kernel(family, update, grouping)
                     .map_err(Error::Invalid)?;
-                let SummaryInputExpr::Column(value) = &update.weight else {
-                    return Err(invalid(
-                        "per-entity update requires a projected value column",
-                    ));
+                let value = match (frequency_item, &update.weight) {
+                    (Some(item), _) => item,
+                    (None, SummaryInputExpr::Column(value)) => value,
+                    (None, _) => {
+                        return Err(invalid(
+                            "per-entity update requires a projected value column",
+                        ))
+                    }
                 };
                 let input = schemas[0].clone();
                 let value = named_column(&input, value)?;
@@ -966,6 +974,27 @@ fn summary_build(
             groups(input, keys)?,
         );
     }
+    // An unkeyed frequency family (UnivMon) counts each value of its
+    // one item column once: a value-frequency build over that column.
+    if let (Some(item), true) = (
+        crate::capability::unit_frequency_column(update),
+        unkeyed_frequency_family(family),
+    ) {
+        crate::capability::validate_summary_kernel(family, update, grouping)
+            .map_err(Error::Invalid)?;
+        let PlannerReduction::Reduce(keys) = reduction else {
+            return Err(invalid(
+                "summary construction requires explicit grouping columns",
+            ));
+        };
+        return Operator::summary_build(
+            input.clone(),
+            family.clone(),
+            named_column(input, item)?,
+            input.time_index,
+            groups(input, keys)?,
+        );
+    }
     if let Some(item) = &update.item {
         let PlannerReduction::Reduce(keys) = reduction else {
             return Err(invalid("keyed summary requires explicit partitions"));
@@ -1028,6 +1057,15 @@ fn summary_build(
         groups(input, keys)?,
     )
 }
+/// Families whose state counts values of one column without keyed items:
+/// their unit-frequency update builds over that column.
+fn unkeyed_frequency_family(family: &FieldDataType) -> bool {
+    matches!(family, FieldDataType::Sketch(kind, _) if matches!(
+        kind.algorithm(),
+        planner_types::post_asap::SketchAlgorithm::UnivMon
+            | planner_types::post_asap::SketchAlgorithm::Hll
+    ))
+}
 fn summary_column(input: &SchemaRef) -> Result<usize, Error> {
     let columns = input
         .fields

crates/asap-physical-operators/src/summary_kernels/factory.rs

@@ -46,6 +46,21 @@ pub trait AccumulatorUpdater: Send {
         }
     }
 
+    /// Update from one typed row value. Numeric kernels keep their Float64
+    /// contract; a frequency kernel may also take non-numeric identities.
+    fn update_value(
+        &mut self,
+        value: &crate::values::Value,
+        timestamp_ms: i64,
+    ) -> Result<(), String> {
+        let crate::values::Value::Float64(value) = value else {
+            return Err("summary update requires Float64".into());
+        };
+        self.validate_single_input(*value)?;
+        self.update_single(*value, timestamp_ms);
+        Ok(())
+    }
+
     /// Feed a single (value, timestamp_ms) pair — for SingleSubpopulation types.
     fn update_single(&mut self, value: f64, timestamp_ms: i64);
 
@@ -645,7 +660,7 @@ pub fn create_planner_accumulator(
         }
     };
     if updater.is_keyed() != input.item.is_some()
-        && !crate::capability::is_unit_sample_frequency(input)
+        && crate::capability::unit_frequency_column(input).is_none()
     {
         return Err("Planner item expression does not match the selected kernel layout".into());
     }
@@ -718,6 +733,11 @@ impl AccumulatorUpdater for HllUpdater {
 }
 
 impl AccumulatorUpdater for UnivMonUpdater {
+    fn update_value(&mut self, value: &crate::values::Value, _: i64) -> Result<(), String> {
+        self.acc
+            .insert_value(value)
+            .map_err(|error| error.to_string())
+    }
     fn is_keyed(&self) -> bool {
         false
     }

crates/asap-physical-operators/src/summary_kernels/univmon.rs

@@ -90,6 +90,27 @@ impl UnivMonAccumulator {
         Ok(())
     }
 
+    /// One occurrence of a typed identity. A finite Float64 is a sample
+    /// ([`Self::insert_sample`]); Bool, Int64 and Utf8 keep their type as
+    /// type-tagged key bytes, since converting Int64 to Float64 would collapse
+    /// neighbouring keys above 2^53. NULL is skipped.
+    pub fn insert_value(&mut self, value: &crate::values::Value) -> Result<(), Error> {
+        use crate::values::Value;
+        match value {
+            Value::Null => return Ok(()),
+            Value::Float64(number) if number.is_finite() => return self.insert_sample(*number),
+            Value::Bool(_) | Value::Int64(_) | Value::Utf8(_) => {}
+            _ => return Err("unsupported UnivMon identity type or non-finite sample".into()),
+        }
+        let key = value.key()?;
+        self.inner
+            .bucket_size
+            .checked_add(1)
+            .ok_or("UnivMon count overflow")?;
+        self.inner.insert(&DataInput::Bytes(&key), 1);
+        Ok(())
+    }
+
     fn compatible(&self, other: &Self) -> bool {
         (
             self.inner.heap_size,
@@ -127,16 +148,28 @@ impl UnivMonAccumulator {
 }
 
 impl AggregateCore for UnivMonAccumulator {
+    /// Counters and heap slots of every layer, plus the bytes of the
+    /// variable-length keys the heaps retain.
     fn approx_memory_bytes(&self) -> usize {
-        std::mem::size_of::<Self>().saturating_add(
-            self.inner.layer_size.saturating_mul(
-                self.inner
-                    .sketch_row
-                    .saturating_mul(self.inner.sketch_col)
-                    .saturating_mul(16)
-                    .saturating_add(self.inner.heap_size.saturating_mul(256)),
-            ),
-        )
+        let key_bytes = (0..self.inner.layer_size)
+            .flat_map(|layer| self.inner.hh_layers[layer].heap())
+            .map(|item| match &item.key {
+                asap_sketchlib::HeapItem::String(key) => key.capacity(),
+                asap_sketchlib::HeapItem::Bytes(key) => key.capacity(),
+                _ => 0,
+            })
+            .fold(0usize, usize::saturating_add);
+        key_bytes
+            .saturating_add(std::mem::size_of::<Self>())
+            .saturating_add(
+                self.inner.layer_size.saturating_mul(
+                    self.inner
+                        .sketch_row
+                        .saturating_mul(self.inner.sketch_col)
+                        .saturating_mul(16)
+                        .saturating_add(self.inner.heap_size.saturating_mul(256)),
+                ),
+            )
     }
     fn clone_boxed_core(&self) -> Box<dyn AggregateCore> {
         Box::new(self.clone())
@@ -248,4 +281,65 @@ mod tests {
         assert!(UnivMonAccumulator::new(4, 11, 2047, 1).is_err());
         assert!(UnivMonAccumulator::new(4, 10, 2047, 1).is_ok());
     }
+
+    // Typed keys stay distinct, and merge, when entered as type-tagged bytes:
+    // Int64 identities above 2^53 do not collapse.
+    #[test]
+    fn typed_keys_merge_and_stay_distinct() {
+        use crate::values::Value;
+        for keys in [
+            [
+                Value::Utf8("192.0.2.1".into()),
+                Value::Utf8("192.0.2.2".into()),
+            ],
+            [
+                Value::Int64(9_007_199_254_740_992),
+                Value::Int64(9_007_199_254_740_993),
+            ],
+            [Value::Bool(false), Value::Bool(true)],
+        ] {
+            let mut left = UnivMonAccumulator::new(32, 5, 1024, 4).unwrap();
+            let mut right = UnivMonAccumulator::new(32, 5, 1024, 4).unwrap();
+            for _ in 0..2 {
+                left.insert_value(&keys[0]).unwrap();
+                right.insert_value(&keys[1]).unwrap();
+            }
+            left.insert_value(&Value::Null).unwrap();
+            left.merge_in_place(&right).unwrap();
+            for (statistic, expected) in [
+                (count(), 4.0),
+                (SketchStatistic::Cardinality, 2.0),
+                (SketchStatistic::FrequencyL2, 8.0_f64.sqrt()),
+                (SketchStatistic::FrequencyEntropy, 1.0),
+            ] {
+                let estimate = left.estimate(&statistic).unwrap();
+                assert!(
+                    (estimate - expected).abs() < 0.01,
+                    "{keys:?} {statistic:?}: {estimate}"
+                );
+            }
+        }
+    }
+
+    // A non-finite sample or an unsupported identity type is an error.
+    #[test]
+    fn rejects_non_finite_and_untyped_identities() {
+        use crate::values::Value;
+        let mut state = UnivMonAccumulator::new(32, 5, 1024, 4).unwrap();
+        assert!(state.insert_value(&Value::Float64(f64::INFINITY)).is_err());
+        assert!(state.insert_value(&Value::Timestamp(1)).is_err());
+    }
+
+    // Retained variable-length identities count toward the state's memory.
+    #[test]
+    fn memory_accounts_for_string_identities() {
+        let mut state = UnivMonAccumulator::new(32, 5, 1024, 4).unwrap();
+        let empty = state.approx_memory_bytes();
+        state
+            .insert_value(&crate::values::Value::Utf8("x".repeat(4096).into()))
+            .unwrap();
+        assert!(state.approx_memory_bytes() >= empty + 4096);
+        state.clear();
+        assert_eq!(state.approx_memory_bytes(), empty);
+    }
 }

crates/asap-physical-operators/src/values.rs

@@ -277,7 +277,7 @@ fn validate_state(family: &SummaryFamilyType, state: &dyn AggregateCore) -> Resu
     use crate::summary_kernels::{
         count_min_sketch::CountMinSketchAccumulator, datasketches_kll::DatasketchesKLLAccumulator,
         dd_sketch::DDSketchAccumulator, exact::ExactAccumulator, hll_sketch::HllSketchAccumulator,
-        kmv::KmvAccumulator,
+        kmv::KmvAccumulator, univmon::UnivMonAccumulator,
     };
     use planner_types::post_asap::SketchParams;
     validate_family(family)?;
@@ -328,6 +328,24 @@ fn validate_state(family: &SummaryFamilyType, state: &dyn AggregateCore) -> Resu
                 .as_any()
                 .downcast_ref::<KmvAccumulator>()
                 .is_some_and(|s| s.inner.k == *k as usize),
+            SketchParams::UnivMon {
+                heap_size,
+                sketch_rows,
+                sketch_cols,
+                layers,
+                ..
+            } => state
+                .as_any()
+                .downcast_ref::<UnivMonAccumulator>()
+                .is_some_and(|s| {
+                    s.dimensions()
+                        == (
+                            *heap_size as usize,
+                            *sketch_rows as usize,
+                            *sketch_cols as usize,
+                            usize::from(*layers),
+                        )
+                }),
             SketchParams::Cms { width, depth } => state
                 .as_any()
                 .downcast_ref::<CountMinSketchAccumulator>()