harmos_domain/
aggregate.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::borrow::Cow;
18// Third Party Imports
19use serde::{Deserialize, Serialize};
20// Local Imports
21use crate::message::{PostcardReader, PostcardValue};
22use crate::schema::declaration::{Field, Kind};
23use crate::schema::{Cardinality, Described, declaration};
24use crate::{MemorySize, MessageError, RowMetadata};
25
26/* === Constants === */
27
28// Protobuf variant tags, hashed from each authored variant name exactly as
29// `#[harmos::message]` derives them; the message tests pin them against a
30// derived twin, so the hand-written codec and the derive cannot drift apart.
31const COUNT: u32 = 31_971_292;
32const SUM: u32 = 534_305_306;
33const MEAN: u32 = 96_322_653;
34const MEDIAN: u32 = 148_660_223;
35const MIN: u32 = 244_600_265;
36const MAX: u32 = 480_766_859;
37const RMS: u32 = 43_272_471;
38const STD: u32 = 450_256_219;
39const PERCENTILE: u32 = 455_936_439;
40const RANK: u32 = 211_531_165;
41
42// Variant tags in declaration order, indexed by the Postcard ordinal.
43
44/* === Macros === */
45
46/* === Type Aliases === */
47
48/* === Structs === */
49
50// The percentile variant payload: one `rank` field carrying the percentage.
51struct Rank;
52
53/* === Enums === */
54
55// The one statistic vocabulary every Harmos reduction answers: plot buckets,
56// storage group-bys and stream batch measurements name what they compute with
57// these members, and `harmos::signal` holds the math behind each of them.
58// `Std` is the population standard deviation. `Percentile` carries a whole
59// percentage, 0 to 100, answered by linear interpolation between ranks.
60//
61// A message field of this type travels as a Harmos enum message, the shape
62// `#[harmos::message]` gives a data-carrying enum: one protobuf oneof whose
63// members are the PascalCase variant names, each holding a payload message.
64// Unit members carry an empty payload; `Percentile` carries `rank` (uint32,
65// refused above 255 on decode). An absent member decodes as `Count`. Postcard
66// writes the declaration ordinal, then the rank byte for `Percentile`, the
67// same bytes serde's Postcard form produces.
68//
69// JSON has two spellings, one per codec:
70// - serde: unit members are snake case strings (`"count"`), a percentile is
71//   `{"percentile": 95}`.
72// - protobuf JSON (Python `MessageToJson`, gRPC gateways):
73//   `{"Count": {}}`, `{"Percentile": {"rank": 95}}`.
74// `name()` and `parse()` keep the compact display spelling (`"p95"`) for
75// labels and text inputs; it is never a wire form.
76#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
77#[serde(rename_all = "snake_case")]
78pub enum Aggregate {
79    Count,
80    Sum,
81    Mean,
82    Median,
83    Min,
84    Max,
85    Rms,
86    Std,
87    Percentile(#[serde(with = "rank")] u8),
88}
89
90/* === Traits === */
91
92/* === Implementations === */
93
94impl Aggregate {
95    // The wire name: the member in snake case, a percentile as `p` and its rank.
96    pub fn name(&self) -> String {
97        match self {
98            Self::Count => "count".to_owned(),
99            Self::Sum => "sum".to_owned(),
100            Self::Mean => "mean".to_owned(),
101            Self::Median => "median".to_owned(),
102            Self::Min => "min".to_owned(),
103            Self::Max => "max".to_owned(),
104            Self::Rms => "rms".to_owned(),
105            Self::Std => "std".to_owned(),
106            Self::Percentile(rank) => format!("p{rank}"),
107        }
108    }
109
110    pub fn parse(name: &str) -> Option<Self> {
111        let named = match name {
112            "count" => Self::Count,
113            "sum" => Self::Sum,
114            "mean" => Self::Mean,
115            "median" => Self::Median,
116            "min" => Self::Min,
117            "max" => Self::Max,
118            "rms" => Self::Rms,
119            "std" => Self::Std,
120            _ => {
121                let rank = name.strip_prefix('p')?;
122                if rank.is_empty() || !rank.bytes().all(|byte| byte.is_ascii_digit()) {
123                    return None;
124                }
125                Self::Percentile(rank.parse().ok()?)
126            }
127        };
128
129        named.answerable().then_some(named)
130    }
131
132    // A percentile rank is a percentage; every other member always answers.
133    pub fn answerable(&self) -> bool {
134        match self {
135            Self::Percentile(rank) => *rank <= 100,
136            _ => true,
137        }
138    }
139
140    // Whether a running summary of count, sum, extremes and spread answers this
141    // member, so partial answers merge without the values. Median and
142    // percentiles need the ordered values themselves.
143    pub fn mergeable(&self) -> bool {
144        !matches!(self, Self::Median | Self::Percentile(_))
145    }
146
147    fn ordinal(&self) -> u32 {
148        match self {
149            Self::Count => 0,
150            Self::Sum => 1,
151            Self::Mean => 2,
152            Self::Median => 3,
153            Self::Min => 4,
154            Self::Max => 5,
155            Self::Rms => 6,
156            Self::Std => 7,
157            Self::Percentile(_) => 8,
158        }
159    }
160
161    fn unit(ordinal: u32) -> Option<Self> {
162        Some(match ordinal {
163            0 => Self::Count,
164            1 => Self::Sum,
165            2 => Self::Mean,
166            3 => Self::Median,
167            4 => Self::Min,
168            5 => Self::Max,
169            6 => Self::Rms,
170            7 => Self::Std,
171            _ => return None,
172        })
173    }
174
175    fn read_postcard(input: &mut PostcardReader<'_>) -> Result<Self, MessageError> {
176        let ordinal: u32 = input.value()?;
177        if ordinal == 8 {
178            // The rank travels as the declared u32 and must fit a percentile rank.
179            let rank: u32 = input.value()?;
180            return u8::try_from(rank).map(Self::Percentile).map_err(|_| MessageError::borrowed("percentile rank exceeds 255"));
181        }
182        Self::unit(ordinal).ok_or(MessageError::borrowed("unknown Postcard enum variant"))
183    }
184}
185
186impl Described for Aggregate {
187    const NAME: &'static str = "harmos::Aggregate";
188    const FIELDS: &'static [Field] = &[
189        variant("Count", COUNT, <()>::DEFINITION),
190        variant("Sum", SUM, <()>::DEFINITION),
191        variant("Mean", MEAN, <()>::DEFINITION),
192        variant("Median", MEDIAN, <()>::DEFINITION),
193        variant("Min", MIN, <()>::DEFINITION),
194        variant("Max", MAX, <()>::DEFINITION),
195        variant("Rms", RMS, <()>::DEFINITION),
196        variant("Std", STD, <()>::DEFINITION),
197        variant("Percentile", PERCENTILE, Rank::DEFINITION),
198    ];
199    const MARKER: &'static [u8] = &declaration::encode::<{ declaration::encoded_len(Self::NAME, Self::FIELDS) }>(Self::NAME, Self::FIELDS);
200}
201
202impl Described for Rank {
203    const NAME: &'static str = "harmos::Aggregate::Percentile";
204    const FIELDS: &'static [Field] = &[Field {
205        name: Cow::Borrowed("rank"),
206        tag: RANK,
207        kind: Kind::U32,
208        cardinality: Cardinality::Required,
209    }];
210    const MARKER: &'static [u8] = &declaration::encode::<{ declaration::encoded_len(Self::NAME, Self::FIELDS) }>(Self::NAME, Self::FIELDS);
211}
212
213impl PostcardValue for Aggregate {
214    fn write(&self, bytes: &mut Vec<u8>) {
215        self.ordinal().write(bytes);
216        if let Self::Percentile(rank) = self {
217            u32::from(*rank).write(bytes);
218        }
219    }
220
221    fn size(&self) -> usize {
222        let rank = match self {
223            Self::Percentile(rank) => u32::from(*rank).size(),
224            _ => 0,
225        };
226        self.ordinal().size().saturating_add(rank)
227    }
228
229    fn memory(input: &mut PostcardReader<'_>) -> Result<usize, MessageError> {
230        input.nested(|input| {
231            Self::read_postcard(input)?;
232            Ok(std::mem::size_of::<Self>())
233        })
234    }
235
236    fn read(input: &mut PostcardReader<'_>) -> Result<Self, MessageError> {
237        input.nested(Self::read_postcard)
238    }
239}
240
241impl MemorySize for Aggregate {
242    fn memory_size(&self) -> usize {
243        std::mem::size_of::<Self>()
244    }
245}
246
247impl RowMetadata for Aggregate {}
248
249/* === Public Functions === */
250
251/* === Private Functions === */
252
253// Serde carries the rank as the declared u32, so its Postcard matches the schema.
254mod rank {
255    pub(super) fn serialize<S: serde::Serializer>(rank: &u8, serializer: S) -> Result<S::Ok, S::Error> {
256        serializer.serialize_u32(u32::from(*rank))
257    }
258
259    pub(super) fn deserialize<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result<u8, D::Error> {
260        let rank = <u32 as serde::Deserialize>::deserialize(deserializer)?;
261        u8::try_from(rank).map_err(|_| serde::de::Error::custom("percentile rank exceeds 255"))
262    }
263}
264
265const fn variant(name: &'static str, tag: u32, payload: declaration::Definition) -> Field {
266    Field {
267        name: Cow::Borrowed(name),
268        tag,
269        kind: Kind::Message { definition: payload },
270        cardinality: Cardinality::Variant,
271    }
272}
273
274// A zero rank is the proto3 default and stays off the wire.
275
276/* === Tests === */
277
278#[cfg(test)]
279mod tests {
280    use super::Aggregate;
281
282    const EVERY: [Aggregate; 11] = [
283        Aggregate::Count,
284        Aggregate::Sum,
285        Aggregate::Mean,
286        Aggregate::Median,
287        Aggregate::Min,
288        Aggregate::Max,
289        Aggregate::Rms,
290        Aggregate::Std,
291        Aggregate::Percentile(0),
292        Aggregate::Percentile(25),
293        Aggregate::Percentile(100),
294    ];
295
296    #[test]
297    fn wire_names() {
298        for kind in EVERY {
299            assert_eq!(Aggregate::parse(&kind.name()), Some(kind));
300            let json = serde_json::to_string(&kind).unwrap();
301            assert_eq!(serde_json::from_str::<Aggregate>(&json).unwrap(), kind);
302        }
303        assert_eq!(serde_json::to_string(&Aggregate::Count).unwrap(), "\"count\"");
304        assert_eq!(serde_json::to_string(&Aggregate::Rms).unwrap(), "\"rms\"");
305        assert_eq!(serde_json::to_string(&Aggregate::Percentile(95)).unwrap(), "{\"percentile\":95}");
306        assert_eq!(Aggregate::Percentile(95).name(), "p95");
307        assert_eq!(Aggregate::parse("p95"), Some(Aggregate::Percentile(95)));
308        for refused in ["p101", "p", "p+5", "variance"] {
309            assert_eq!(Aggregate::parse(refused), None, "{refused}");
310        }
311        assert!(!Aggregate::Percentile(101).answerable());
312    }
313}
harmos_domain/aggregate.rs

harmos_domain/
aggregate.rs
