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, kept exactly up to
326    /// [`SKIPPED_EXACT`] names so the count is exact for ordinary records.
327    skipped: std::collections::HashSet<String>,
328    /// Every rejected name, as a bounded sketch: records keyed by data (an id
329    /// or a date per key) would otherwise grow the exact set without limit.
330    skipped_sketch: HyperLogLog,
331}
332
333/// How many rejected column names are remembered exactly before the count
334/// switches to the bounded estimate.
335const SKIPPED_EXACT: usize = 1024;
336
337impl Profiler {
338    pub fn new(spec: ProfilingSpec) -> Self {
339        Self {
340            spec,
341            rows: 0,
342            columns: HashMap::new(),
343            skipped: std::collections::HashSet::new(),
344            skipped_sketch: HyperLogLog::new(),
345        }
346    }
347
348    pub fn spec(&self) -> &ProfilingSpec {
349        &self.spec
350    }
351
352    /// Records observed so far.
353    pub fn rows(&self) -> u64 {
354        self.rows
355    }
356
357    /// Fold one written page in. Non-object records count toward `rows` but
358    /// contribute no columns.
359    pub fn observe_page(&mut self, records: &[Value]) {
360        for record in records {
361            self.rows += 1;
362            let Some(obj) = record.as_object() else {
363                continue;
364            };
365            for (name, v) in obj {
366                if !self.spec.selects(name) {
367                    continue;
368                }
369                if let Some(acc) = self.columns.get_mut(name) {
370                    acc.observe(v);
371                    continue;
372                }
373                if self.columns.len() >= self.spec.max_columns {
374                    self.skipped_sketch.insert(name.as_str());
375                    if self.skipped.len() < SKIPPED_EXACT && !self.skipped.contains(name) {
376                        self.skipped.insert(name.clone());
377                    }
378                    continue;
379                }
380                let mut acc =
381                    ColumnAcc::new(self.spec.top_values, self.spec.categorical_max_distinct);
382                acc.observe(v);
383                self.columns.insert(name.clone(), acc);
384            }
385        }
386    }
387
388    /// The run's profile from everything observed so far.
389    pub fn finish(&self) -> RunProfile {
390        RunProfile {
391            rows: self.rows,
392            columns: self
393                .columns
394                .iter()
395                .map(|(name, acc)| (name.clone(), acc.finish(self.rows, &self.spec)))
396                .collect(),
397            skipped_columns: if self.skipped.len() < SKIPPED_EXACT {
398                self.skipped.len() as u64
399            } else {
400                self.skipped_sketch.estimate().max(SKIPPED_EXACT as u64)
401            },
402        }
403    }
404}
405
406#[cfg(test)]
407mod tests {
408    use super::*;
409    use serde_json::json;
410
411    /// #862 L-034: an object's text does not depend on key insertion order.
412    #[test]
413    fn value_text_is_independent_of_key_order() {
414        let mut a = serde_json::Map::new();
415        a.insert("b".into(), json!(1));
416        a.insert("a".into(), json!(2));
417        let mut b = serde_json::Map::new();
418        b.insert("a".into(), json!(2));
419        b.insert("b".into(), json!(1));
420        assert_eq!(value_text(&Value::Object(a)), r#"{"a":2,"b":1}"#);
421        assert_eq!(value_text(&Value::Object(b)), r#"{"a":2,"b":1}"#);
422    }
423
424    fn profile(records: &[Value]) -> RunProfile {
425        let mut p = Profiler::new(ProfilingSpec::default());
426        p.observe_page(records);
427        p.finish()
428    }
429
430    #[test]
431    fn null_rate_counts_missing_and_explicit_nulls() {
432        let rp = profile(&[
433            json!({"a": 1}),
434            json!({"a": null}),
435            json!({}),
436            json!({"a": 2}),
437        ]);
438        let a = &rp.columns["a"];
439        assert_eq!(rp.rows, 4);
440        assert_eq!(a.present, 3);
441        assert_eq!(a.nulls, 2);
442        assert_eq!(a.null_rate, 0.5);
443        assert_eq!(a.non_null(), 2);
444        assert_eq!(a.types.null, 1);
445        assert_eq!(a.types.integer, 2);
446        assert_eq!(a.types.dominant(), Some("integer"));
447    }
448
449    #[test]
450    fn numeric_string_and_type_mix_summaries() {
451        let rp = profile(&[
452            json!({"n": 1, "s": "ab", "b": true, "o": {"x": 1}, "l": [1]}),
453            json!({"n": 2.5, "s": "abcd", "b": false, "o": {"x": 2}, "l": [1, 2]}),
454            json!({"n": 3, "s": "", "b": true, "o": {"x": 1}, "l": [1]}),
455        ]);
456        let n = rp.columns["n"].numeric.as_ref().unwrap();
457        assert_eq!(n.count, 3);
458        assert_eq!(n.min, 1.0);
459        assert_eq!(n.max, 3.0);
460        assert!((n.mean - 2.1666).abs() < 1e-3);
461        assert_eq!(n.p50, 2.5);
462        assert_eq!(rp.columns["n"].types.integer, 2);
463        assert_eq!(rp.columns["n"].types.number, 1);
464        let s = rp.columns["s"].string.as_ref().unwrap();
465        assert_eq!((s.len_min, s.len_max), (0, 4));
466        assert_eq!(s.len_mean, 2.0);
467        assert!(rp.columns["s"].numeric.is_none());
468        assert_eq!(rp.columns["b"].types.boolean, 3);
469        assert_eq!(rp.columns["o"].types.object, 3);
470        assert_eq!(rp.columns["o"].distinct, 2);
471        assert_eq!(rp.columns["l"].types.array, 3);
472        assert_eq!(
473            rp.columns["b"].types.present(),
474            vec![("boolean", 3)],
475            "present lists only non-zero types"
476        );
477    }
478
479    #[test]
480    fn top_values_with_shares_and_high_cardinality_cutoff() {
481        let mut rows = Vec::new();
482        for i in 0..100 {
483            let c = if i % 2 == 0 {
484                "eu"
485            } else if i % 3 == 0 {
486                "us"
487            } else {
488                "apac"
489            };
490            rows.push(json!({"region": c, "id": format!("id-{i}"), "n": null}));
491        }
492        let spec = ProfilingSpec {
493            categorical_max_distinct: 20,
494            ..Default::default()
495        };
496        let mut p = Profiler::new(spec);
497        p.observe_page(&rows);
498        let rp = p.finish();
499        let region = &rp.columns["region"];
500        let top = region.top_values.as_ref().unwrap();
501        assert_eq!(top[0].value, "eu");
502        assert_eq!(top[0].count, 50);
503        assert_eq!(top[0].share, 0.5);
504        assert!(!region.high_cardinality);
505        let id = &rp.columns["id"];
506        assert!(id.high_cardinality);
507        assert!(id.top_values.is_none());
508        assert!(id.distinct >= 95 && id.distinct <= 105, "{}", id.distinct);
509        // An all-null column has no values to rank.
510        assert!(rp.columns["n"].top_values.is_none());
511        assert_eq!(rp.columns["n"].null_rate, 1.0);
512    }
513
514    #[test]
515    fn top_values_zero_disables_the_sketch() {
516        let spec = ProfilingSpec {
517            top_values: 0,
518            ..Default::default()
519        };
520        let mut p = Profiler::new(spec);
521        p.observe_page(&[json!({"a": "x"})]);
522        assert!(p.finish().columns["a"].top_values.is_none());
523    }
524
525    #[test]
526    fn rejected_column_names_are_counted_in_bounded_memory() {
527        let mut p = Profiler::new(ProfilingSpec {
528            max_columns: 1,
529            ..Default::default()
530        });
531        let page: Vec<Value> = (0..20_000)
532            .map(|i| json!({"id": i, format!("k{i}"): 1}))
533            .collect();
534        p.observe_page(&page);
535        assert_eq!(
536            p.skipped.len(),
537            SKIPPED_EXACT,
538            "the exact set stops growing"
539        );
540        let n = p.finish().skipped_columns;
541        assert!((19_000..=21_000).contains(&n), "estimate {n} ~ 20000");
542    }
543
544    #[test]
545    fn selection_cap_and_non_object_records() {
546        let mut spec = ProfilingSpec {
547            max_columns: 2,
548            ..Default::default()
549        };
550        spec.exclude.push("skip".into());
551        let mut p = Profiler::new(spec);
552        p.observe_page(&[
553            json!({"a": 1, "b": 2, "c": 3, "d": 4, "skip": 5, "_faucet_run_id": "r"}),
554            json!("scalar"),
555            json!({"a": 1, "c": 3, "e": 9}),
556        ]);
557        let rp = p.finish();
558        assert_eq!(rp.rows, 3);
559        assert_eq!(rp.columns.len(), 2);
560        assert!(rp.columns.contains_key("a") && rp.columns.contains_key("b"));
561        assert_eq!(
562            rp.skipped_columns, 3,
563            "c, d, e were capped; skip/_faucet_ excluded"
564        );
565        assert!(rp.truncated());
566        assert_eq!(p.rows(), 3);
567        assert_eq!(p.spec().max_columns, 2);
568    }
569
570    #[test]
571    fn strings_and_numbers_are_distinct_values_and_long_text_is_truncated() {
572        let long = "x".repeat(600);
573        let rp = profile(&[json!({"v": 1}), json!({"v": "1"}), json!({"v": long})]);
574        let v = &rp.columns["v"];
575        assert_eq!(v.distinct, 3);
576        let top = v.top_values.as_ref().unwrap();
577        let truncated = top.iter().find(|t| t.value.ends_with('…')).unwrap();
578        assert_eq!(truncated.value.chars().count(), MAX_VALUE_TEXT + 1);
579    }
580
581    #[test]
582    fn profile_round_trips_through_json() {
583        let rp = profile(&[json!({"a": 1.5, "b": "x"}), json!({"a": null})]);
584        let text = serde_json::to_string(&rp).unwrap();
585        let back: RunProfile = serde_json::from_str(&text).unwrap();
586        assert_eq!(back, rp);
587        assert!(!text.contains("skipped_columns"), "zero is elided: {text}");
588        assert!(!text.contains("\"error\""));
589    }
590
591    #[test]
592    fn empty_run_profile() {
593        let rp = profile(&[]);
594        assert_eq!(rp.rows, 0);
595        assert!(rp.columns.is_empty());
596        assert!(!rp.truncated());
597        assert_eq!(TypeCounts::default().dominant(), None);
598    }
599
600    #[test]
601    fn a_high_cardinality_column_stops_feeding_its_top_values() {
602        let mut acc = ColumnAcc::new(10, 100);
603        for i in 0..1024 {
604            acc.observe(&serde_json::json!(format!("id-{i}")));
605        }
606        assert!(acc.top.is_none());
607        let mut low = ColumnAcc::new(10, 100);
608        for i in 0..1024 {
609            low.observe(&serde_json::json!(format!("v{}", i % 5)));
610        }
611        assert!(low.top.is_some());
612    }
613}
faucet_core/profiling/column.rs

faucet_core/profiling/
column.rs
