faucet_core/profiling/
column.rs

1//! Per-run column profiling (#708): the accumulator that turns written pages
2//! into one [`RunProfile`] — a [`ColumnProfile`] per top-level column with
3//! null rate, type mix, distinct estimate, numeric / string summaries and
4//! the most frequent values. Nested objects and arrays count as one column
5//! (their canonical JSON is the value), matching schema-drift semantics.
6
7use super::sketch::{HyperLogLog, Reservoir, TopK, Welford, hash64};
8use super::spec::ProfilingSpec;
9use schemars::JsonSchema;
10use serde::{Deserialize, Serialize};
11use serde_json::Value;
12use std::collections::{BTreeMap, HashMap};
13
14/// Reservoir size for approximate quantiles.
15const RESERVOIR_CAPACITY: usize = 1024;
16/// Longest value text kept in the top-values sketch (bounds memory per
17/// counter; a longer value is truncated with a marker).
18const MAX_VALUE_TEXT: usize = 256;
19
20/// How many values of each JSON type the column held.
21#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
22#[serde(default)]
23pub struct TypeCounts {
24    pub null: u64,
25    pub boolean: u64,
26    pub integer: u64,
27    pub number: u64,
28    pub string: u64,
29    pub object: u64,
30    pub array: u64,
31}
32
33impl TypeCounts {
34    /// `(type name, count)` for every non-zero type, fixed order.
35    pub fn present(&self) -> Vec<(&'static str, u64)> {
36        [
37            ("null", self.null),
38            ("boolean", self.boolean),
39            ("integer", self.integer),
40            ("number", self.number),
41            ("string", self.string),
42            ("object", self.object),
43            ("array", self.array),
44        ]
45        .into_iter()
46        .filter(|(_, n)| *n > 0)
47        .collect()
48    }
49
50    /// The most common non-null type, if any value was non-null.
51    pub fn dominant(&self) -> Option<&'static str> {
52        self.present()
53            .into_iter()
54            .filter(|(t, _)| *t != "null")
55            .max_by_key(|(_, n)| *n)
56            .map(|(t, _)| t)
57    }
58}
59
60/// Summary of the numeric values a column held.
61#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
62pub struct NumericSummary {
63    pub count: u64,
64    pub min: f64,
65    pub max: f64,
66    pub mean: f64,
67    pub stddev: f64,
68    /// Approximate median (reservoir sample).
69    pub p50: f64,
70    /// Approximate 95th percentile (reservoir sample).
71    pub p95: f64,
72}
73
74/// Summary of the string values a column held (character lengths).
75#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
76pub struct StringSummary {
77    pub count: u64,
78    pub len_min: u64,
79    pub len_max: u64,
80    pub len_mean: f64,
81}
82
83/// One of a column's most frequent values.
84#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
85pub struct TopValue {
86    pub value: String,
87    /// Occurrences (an upper bound; `error` is the possible over-count).
88    pub count: u64,
89    #[serde(default, skip_serializing_if = "is_zero")]
90    pub error: u64,
91    /// `count / non-null values` in this run.
92    pub share: f64,
93}
94
95fn is_zero(n: &u64) -> bool {
96    *n == 0
97}
98
99/// The learned profile of one column over one run.
100#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
101pub struct ColumnProfile {
102    /// Records the run wrote (the denominator of `null_rate`).
103    pub rows: u64,
104    /// Records in which the column was present (including explicit nulls).
105    pub present: u64,
106    /// Records where the column was absent or null.
107    pub nulls: u64,
108    /// `nulls / rows` (0 when no rows).
109    pub null_rate: f64,
110    pub types: TypeCounts,
111    /// Estimated distinct non-null values (HyperLogLog, ~1.6 % error).
112    pub distinct: u64,
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub numeric: Option<NumericSummary>,
115    #[serde(default, skip_serializing_if = "Option::is_none")]
116    pub string: Option<StringSummary>,
117    /// Most frequent values, count descending. Absent for high-cardinality
118    /// columns and when `top_values: 0`.
119    #[serde(default, skip_serializing_if = "Option::is_none")]
120    pub top_values: Option<Vec<TopValue>>,
121    /// The estimated distinct count exceeded `categorical_max_distinct`, so
122    /// no values are published for this column.
123    #[serde(default)]
124    pub high_cardinality: bool,
125}
126
127impl ColumnProfile {
128    /// Non-null values observed.
129    pub fn non_null(&self) -> u64 {
130        self.rows.saturating_sub(self.nulls)
131    }
132}
133
134/// The profile of everything one run wrote.
135#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
136pub struct RunProfile {
137    /// Records the run wrote.
138    pub rows: u64,
139    /// Profiled columns, name-ordered.
140    pub columns: BTreeMap<String, ColumnProfile>,
141    /// Columns seen beyond `max_columns` and therefore not profiled.
142    #[serde(default, skip_serializing_if = "is_zero")]
143    pub skipped_columns: u64,
144}
145
146impl RunProfile {
147    /// Whether the column cap was hit.
148    pub fn truncated(&self) -> bool {
149        self.skipped_columns > 0
150    }
151}
152
153/// How often (in observed values) a column's distinct estimate is compared
154/// with `categorical_max_distinct`.
155const CARDINALITY_CHECK_EVERY: u64 = 256;
156
157/// Per-column accumulator.
158#[derive(Debug)]
159struct ColumnAcc {
160    present: u64,
161    nulls: u64,
162    types: TypeCounts,
163    distinct: HyperLogLog,
164    top: Option<TopK>,
165    /// `categorical_max_distinct`: once the distinct estimate passes it the
166    /// column reports no top values, so the sketch stops being fed.
167    cardinality_cap: u64,
168    numeric: Welford,
169    reservoir: Reservoir,
170    string_len: Welford,
171}
172
173impl ColumnAcc {
174    fn new(top_values: usize, cardinality_cap: u64) -> Self {
175        Self {
176            present: 0,
177            nulls: 0,
178            types: TypeCounts::default(),
179            distinct: HyperLogLog::new(),
180            top: (top_values > 0).then(|| TopK::new(top_values)),
181            cardinality_cap,
182            numeric: Welford::default(),
183            reservoir: Reservoir::new(RESERVOIR_CAPACITY),
184            string_len: Welford::default(),
185        }
186    }
187
188    fn observe(&mut self, v: &Value) {
189        self.present += 1;
190        match v {
191            Value::Null => {
192                self.nulls += 1;
193                self.types.null += 1;
194                return;
195            }
196            Value::Bool(_) => self.types.boolean += 1,
197            Value::Number(n) => {
198                if n.is_i64() || n.is_u64() {
199                    self.types.integer += 1;
200                } else {
201                    self.types.number += 1;
202                }
203                if let Some(f) = n.as_f64()
204                    && f.is_finite()
205                {
206                    self.numeric.insert(f);
207                    self.reservoir.insert(f);
208                }
209            }
210            Value::String(s) => {
211                self.types.string += 1;
212                self.string_len.insert(s.chars().count() as f64);
213            }
214            Value::Object(_) => self.types.object += 1,
215            Value::Array(_) => self.types.array += 1,
216        }
217        let text = value_text(v);
218        // Distinctness is typed — `1` and `"1"` are two values — while the
219        // frequency sketch keys on the display text.
220        self.distinct
221            .insert_hash(hash64(&(type_tag(v), text.as_ref())));
222        if let Some(top) = &mut self.top {
223            top.insert(&truncate_text(&text));
224            // The estimate only grows, so a column clearly past the cap (a
225            // margin over the sketch's error) ends the run past it too and
226            // will report no top values.
227            if self.present.is_multiple_of(CARDINALITY_CHECK_EVERY)
228                && self.distinct.estimate() > self.cardinality_cap.saturating_mul(9) / 8 + 16
229            {
230                self.top = None;
231            }
232        }
233    }
234
235    fn finish(&self, rows: u64, spec: &ProfilingSpec) -> ColumnProfile {
236        let missing = rows.saturating_sub(self.present);
237        let nulls = self.nulls + missing;
238        let non_null = rows.saturating_sub(nulls);
239        let distinct = self.distinct.estimate();
240        let high_cardinality = distinct > spec.categorical_max_distinct;
241        let numeric = (self.numeric.count > 0).then(|| NumericSummary {
242            count: self.numeric.count,
243            min: self.numeric.min,
244            max: self.numeric.max,
245            mean: self.numeric.mean().unwrap_or(0.0),
246            stddev: self.numeric.stddev().unwrap_or(0.0),
247            p50: self.reservoir.quantile(0.5).unwrap_or(0.0),
248            p95: self.reservoir.quantile(0.95).unwrap_or(0.0),
249        });
250        let string = (self.string_len.count > 0).then(|| StringSummary {
251            count: self.string_len.count,
252            len_min: self.string_len.min as u64,
253            len_max: self.string_len.max as u64,
254            len_mean: self.string_len.mean().unwrap_or(0.0),
255        });
256        let top_values = match &self.top {
257            Some(top) if !high_cardinality && non_null > 0 => Some(
258                top.top(spec.top_values)
259                    .into_iter()
260                    .map(|(value, count, error)| TopValue {
261                        value,
262                        count,
263                        error,
264                        share: count as f64 / non_null as f64,
265                    })
266                    .collect(),
267            ),
268            _ => None,
269        };
270        ColumnProfile {
271            rows,
272            present: self.present,
273            nulls,
274            null_rate: if rows == 0 {
275                0.0
276            } else {
277                nulls as f64 / rows as f64
278            },
279            types: self.types,
280            distinct,
281            numeric,
282            string,
283            top_values,
284            high_cardinality,
285        }
286    }
287}
288
289/// The text a value is counted by: strings verbatim, everything else as
290/// canonical JSON (so `1` and `"1"` are distinct values).
291fn value_text(v: &Value) -> std::borrow::Cow<'_, str> {
292    match v {
293        Value::String(s) => std::borrow::Cow::Borrowed(s.as_str()),
294        other => std::borrow::Cow::Owned(crate::util::canonical_json(other)),
295    }
296}
297
298fn type_tag(v: &Value) -> u8 {
299    match v {
300        Value::Null => 0,
301        Value::Bool(_) => 1,
302        Value::Number(_) => 2,
303        Value::String(_) => 3,
304        Value::Array(_) => 4,
305        Value::Object(_) => 5,
306    }
307}
308
309fn truncate_text(s: &str) -> String {
310    if s.chars().count() <= MAX_VALUE_TEXT {
311        return s.to_string();
312    }
313    let mut out: String = s.chars().take(MAX_VALUE_TEXT).collect();
314    out.push('…');
315    out
316}
317
318/// The run-level accumulator: hand it every page the sink wrote, then
319/// [`finish`](Profiler::finish).
320#[derive(Debug)]
321pub struct Profiler {
322    spec: ProfilingSpec,
323    rows: u64,
324    columns: HashMap<String, ColumnAcc>,
325    /// Column names seen and rejected by the cap (so the count is exact).
326    skipped: std::collections::HashSet<String>,
327}
328
329impl Profiler {
330    pub fn new(spec: ProfilingSpec) -> Self {
331        Self {
332            spec,
333            rows: 0,
334            columns: HashMap::new(),
335            skipped: std::collections::HashSet::new(),
336        }
337    }
338
339    pub fn spec(&self) -> &ProfilingSpec {
340        &self.spec
341    }
342
343    /// Records observed so far.
344    pub fn rows(&self) -> u64 {
345        self.rows
346    }
347
348    /// Fold one written page in. Non-object records count toward `rows` but
349    /// contribute no columns.
350    pub fn observe_page(&mut self, records: &[Value]) {
351        for record in records {
352            self.rows += 1;
353            let Some(obj) = record.as_object() else {
354                continue;
355            };
356            for (name, v) in obj {
357                if !self.spec.selects(name) {
358                    continue;
359                }
360                if let Some(acc) = self.columns.get_mut(name) {
361                    acc.observe(v);
362                    continue;
363                }
364                if self.columns.len() >= self.spec.max_columns {
365                    self.skipped.insert(name.clone());
366                    continue;
367                }
368                let mut acc =
369                    ColumnAcc::new(self.spec.top_values, self.spec.categorical_max_distinct);
370                acc.observe(v);
371                self.columns.insert(name.clone(), acc);
372            }
373        }
374    }
375
376    /// The run's profile from everything observed so far.
377    pub fn finish(&self) -> RunProfile {
378        RunProfile {
379            rows: self.rows,
380            columns: self
381                .columns
382                .iter()
383                .map(|(name, acc)| (name.clone(), acc.finish(self.rows, &self.spec)))
384                .collect(),
385            skipped_columns: self.skipped.len() as u64,
386        }
387    }
388}
389
390#[cfg(test)]
391mod tests {
392    use super::*;
393    use serde_json::json;
394
395    /// #862 L-034: an object's text does not depend on key insertion order.
396    #[test]
397    fn value_text_is_independent_of_key_order() {
398        let mut a = serde_json::Map::new();
399        a.insert("b".into(), json!(1));
400        a.insert("a".into(), json!(2));
401        let mut b = serde_json::Map::new();
402        b.insert("a".into(), json!(2));
403        b.insert("b".into(), json!(1));
404        assert_eq!(value_text(&Value::Object(a)), r#"{"a":2,"b":1}"#);
405        assert_eq!(value_text(&Value::Object(b)), r#"{"a":2,"b":1}"#);
406    }
407
408    fn profile(records: &[Value]) -> RunProfile {
409        let mut p = Profiler::new(ProfilingSpec::default());
410        p.observe_page(records);
411        p.finish()
412    }
413
414    #[test]
415    fn null_rate_counts_missing_and_explicit_nulls() {
416        let rp = profile(&[
417            json!({"a": 1}),
418            json!({"a": null}),
419            json!({}),
420            json!({"a": 2}),
421        ]);
422        let a = &rp.columns["a"];
423        assert_eq!(rp.rows, 4);
424        assert_eq!(a.present, 3);
425        assert_eq!(a.nulls, 2);
426        assert_eq!(a.null_rate, 0.5);
427        assert_eq!(a.non_null(), 2);
428        assert_eq!(a.types.null, 1);
429        assert_eq!(a.types.integer, 2);
430        assert_eq!(a.types.dominant(), Some("integer"));
431    }
432
433    #[test]
434    fn numeric_string_and_type_mix_summaries() {
435        let rp = profile(&[
436            json!({"n": 1, "s": "ab", "b": true, "o": {"x": 1}, "l": [1]}),
437            json!({"n": 2.5, "s": "abcd", "b": false, "o": {"x": 2}, "l": [1, 2]}),
438            json!({"n": 3, "s": "", "b": true, "o": {"x": 1}, "l": [1]}),
439        ]);
440        let n = rp.columns["n"].numeric.as_ref().unwrap();
441        assert_eq!(n.count, 3);
442        assert_eq!(n.min, 1.0);
443        assert_eq!(n.max, 3.0);
444        assert!((n.mean - 2.1666).abs() < 1e-3);
445        assert_eq!(n.p50, 2.5);
446        assert_eq!(rp.columns["n"].types.integer, 2);
447        assert_eq!(rp.columns["n"].types.number, 1);
448        let s = rp.columns["s"].string.as_ref().unwrap();
449        assert_eq!((s.len_min, s.len_max), (0, 4));
450        assert_eq!(s.len_mean, 2.0);
451        assert!(rp.columns["s"].numeric.is_none());
452        assert_eq!(rp.columns["b"].types.boolean, 3);
453        assert_eq!(rp.columns["o"].types.object, 3);
454        assert_eq!(rp.columns["o"].distinct, 2);
455        assert_eq!(rp.columns["l"].types.array, 3);
456        assert_eq!(
457            rp.columns["b"].types.present(),
458            vec![("boolean", 3)],
459            "present lists only non-zero types"
460        );
461    }
462
463    #[test]
464    fn top_values_with_shares_and_high_cardinality_cutoff() {
465        let mut rows = Vec::new();
466        for i in 0..100 {
467            let c = if i % 2 == 0 {
468                "eu"
469            } else if i % 3 == 0 {
470                "us"
471            } else {
472                "apac"
473            };
474            rows.push(json!({"region": c, "id": format!("id-{i}"), "n": null}));
475        }
476        let spec = ProfilingSpec {
477            categorical_max_distinct: 20,
478            ..Default::default()
479        };
480        let mut p = Profiler::new(spec);
481        p.observe_page(&rows);
482        let rp = p.finish();
483        let region = &rp.columns["region"];
484        let top = region.top_values.as_ref().unwrap();
485        assert_eq!(top[0].value, "eu");
486        assert_eq!(top[0].count, 50);
487        assert_eq!(top[0].share, 0.5);
488        assert!(!region.high_cardinality);
489        let id = &rp.columns["id"];
490        assert!(id.high_cardinality);
491        assert!(id.top_values.is_none());
492        assert!(id.distinct >= 95 && id.distinct <= 105, "{}", id.distinct);
493        // An all-null column has no values to rank.
494        assert!(rp.columns["n"].top_values.is_none());
495        assert_eq!(rp.columns["n"].null_rate, 1.0);
496    }
497
498    #[test]
499    fn top_values_zero_disables_the_sketch() {
500        let spec = ProfilingSpec {
501            top_values: 0,
502            ..Default::default()
503        };
504        let mut p = Profiler::new(spec);
505        p.observe_page(&[json!({"a": "x"})]);
506        assert!(p.finish().columns["a"].top_values.is_none());
507    }
508
509    #[test]
510    fn selection_cap_and_non_object_records() {
511        let mut spec = ProfilingSpec {
512            max_columns: 2,
513            ..Default::default()
514        };
515        spec.exclude.push("skip".into());
516        let mut p = Profiler::new(spec);
517        p.observe_page(&[
518            json!({"a": 1, "b": 2, "c": 3, "d": 4, "skip": 5, "_faucet_run_id": "r"}),
519            json!("scalar"),
520            json!({"a": 1, "c": 3, "e": 9}),
521        ]);
522        let rp = p.finish();
523        assert_eq!(rp.rows, 3);
524        assert_eq!(rp.columns.len(), 2);
525        assert!(rp.columns.contains_key("a") && rp.columns.contains_key("b"));
526        assert_eq!(
527            rp.skipped_columns, 3,
528            "c, d, e were capped; skip/_faucet_ excluded"
529        );
530        assert!(rp.truncated());
531        assert_eq!(p.rows(), 3);
532        assert_eq!(p.spec().max_columns, 2);
533    }
534
535    #[test]
536    fn strings_and_numbers_are_distinct_values_and_long_text_is_truncated() {
537        let long = "x".repeat(600);
538        let rp = profile(&[json!({"v": 1}), json!({"v": "1"}), json!({"v": long})]);
539        let v = &rp.columns["v"];
540        assert_eq!(v.distinct, 3);
541        let top = v.top_values.as_ref().unwrap();
542        let truncated = top.iter().find(|t| t.value.ends_with('…')).unwrap();
543        assert_eq!(truncated.value.chars().count(), MAX_VALUE_TEXT + 1);
544    }
545
546    #[test]
547    fn profile_round_trips_through_json() {
548        let rp = profile(&[json!({"a": 1.5, "b": "x"}), json!({"a": null})]);
549        let text = serde_json::to_string(&rp).unwrap();
550        let back: RunProfile = serde_json::from_str(&text).unwrap();
551        assert_eq!(back, rp);
552        assert!(!text.contains("skipped_columns"), "zero is elided: {text}");
553        assert!(!text.contains("\"error\""));
554    }
555
556    #[test]
557    fn empty_run_profile() {
558        let rp = profile(&[]);
559        assert_eq!(rp.rows, 0);
560        assert!(rp.columns.is_empty());
561        assert!(!rp.truncated());
562        assert_eq!(TypeCounts::default().dominant(), None);
563    }
564
565    #[test]
566    fn a_high_cardinality_column_stops_feeding_its_top_values() {
567        let mut acc = ColumnAcc::new(10, 100);
568        for i in 0..1024 {
569            acc.observe(&serde_json::json!(format!("id-{i}")));
570        }
571        assert!(acc.top.is_none());
572        let mut low = ColumnAcc::new(10, 100);
573        for i in 0..1024 {
574            low.observe(&serde_json::json!(format!("v{}", i % 5)));
575        }
576        assert!(low.top.is_some());
577    }
578}
faucet_core/profiling/column.rs

faucet_core/profiling/
column.rs
