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

harmos_domain/
aggregate.rs
