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] =
205        &[Field { name: Cow::Borrowed("rank"), tag: RANK, kind: Kind::U32, cardinality: Cardinality::Required }];
206    const MARKER: &'static [u8] = &declaration::encode::<{ declaration::encoded_len(Self::NAME, Self::FIELDS) }>(Self::NAME, Self::FIELDS);
207}
208
209impl PostcardValue for Aggregate {
210    fn write(&self, bytes: &mut Vec<u8>) {
211        self.ordinal().write(bytes);
212        if let Self::Percentile(rank) = self {
213            u32::from(*rank).write(bytes);
214        }
215    }
216
217    fn size(&self) -> usize {
218        let rank = match self {
219            Self::Percentile(rank) => u32::from(*rank).size(),
220            _ => 0,
221        };
222        self.ordinal().size().saturating_add(rank)
223    }
224
225    fn memory(input: &mut PostcardReader<'_>) -> Result<usize, MessageError> {
226        input.nested(|input| {
227            Self::read_postcard(input)?;
228            Ok(std::mem::size_of::<Self>())
229        })
230    }
231
232    fn read(input: &mut PostcardReader<'_>) -> Result<Self, MessageError> {
233        input.nested(Self::read_postcard)
234    }
235}
236
237impl MemorySize for Aggregate {
238    fn memory_size(&self) -> usize {
239        std::mem::size_of::<Self>()
240    }
241}
242
243impl RowMetadata for Aggregate {}
244
245/* === Public Functions === */
246
247/* === Private Functions === */
248
249// Serde carries the rank as the declared u32, so its Postcard matches the schema.
250mod rank {
251    pub(super) fn serialize<S: serde::Serializer>(rank: &u8, serializer: S) -> Result<S::Ok, S::Error> {
252        serializer.serialize_u32(u32::from(*rank))
253    }
254
255    pub(super) fn deserialize<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result<u8, D::Error> {
256        let rank = <u32 as serde::Deserialize>::deserialize(deserializer)?;
257        u8::try_from(rank).map_err(|_| serde::de::Error::custom("percentile rank exceeds 255"))
258    }
259}
260
261const fn variant(name: &'static str, tag: u32, payload: declaration::Definition) -> Field {
262    Field { name: Cow::Borrowed(name), tag, kind: Kind::Message { definition: payload }, cardinality: Cardinality::Variant }
263}
264
265// A zero rank is the proto3 default and stays off the wire.
266
267/* === Tests === */
268
269#[cfg(test)]
270mod tests {
271    use super::Aggregate;
272
273    const EVERY: [Aggregate; 11] = [
274        Aggregate::Count,
275        Aggregate::Sum,
276        Aggregate::Mean,
277        Aggregate::Median,
278        Aggregate::Min,
279        Aggregate::Max,
280        Aggregate::Rms,
281        Aggregate::Std,
282        Aggregate::Percentile(0),
283        Aggregate::Percentile(25),
284        Aggregate::Percentile(100),
285    ];
286
287    #[test]
288    fn wire_names() {
289        for kind in EVERY {
290            assert_eq!(Aggregate::parse(&kind.name()), Some(kind));
291            let json = serde_json::to_string(&kind).unwrap();
292            assert_eq!(serde_json::from_str::<Aggregate>(&json).unwrap(), kind);
293        }
294        assert_eq!(serde_json::to_string(&Aggregate::Count).unwrap(), "\"count\"");
295        assert_eq!(serde_json::to_string(&Aggregate::Rms).unwrap(), "\"rms\"");
296        assert_eq!(serde_json::to_string(&Aggregate::Percentile(95)).unwrap(), "{\"percentile\":95}");
297        assert_eq!(Aggregate::Percentile(95).name(), "p95");
298        assert_eq!(Aggregate::parse("p95"), Some(Aggregate::Percentile(95)));
299        for refused in ["p101", "p", "p+5", "variance"] {
300            assert_eq!(Aggregate::parse(refused), None, "{refused}");
301        }
302        assert!(!Aggregate::Percentile(101).answerable());
303    }
304}
harmos_domain/aggregate.rs

harmos_domain/
aggregate.rs
