feat(stats): fluree-db-stats kernel and on-demand column profiling - #1789
feat(stats): fluree-db-stats kernel and on-demand column profiling#1789bplatz wants to merge 2 commits into
Conversation
…iling New crate `fluree-db-stats` holds mergeable sketches shared by every place that summarises a column of values: a const-generic HyperLogLog, exact Welford/Chan moments, a merging t-digest, and Misra-Gries heavy hitters, composed into `ColumnProfile` and a bounded-group `GroupedProfile`. The `findings` module derives group baselines, concentration, per-group constancy and scale drift for quality rules; the `tabular` feature adds a face over `fluree-db-tabular` column batches. The indexer's `HllSketch256` is now a re-export of the kernel's `Hll256`. Register layout and the 256-byte persisted form are unchanged, so existing sketches load and merge as before. `fluree-db-api` gains `Fluree::profile_ledger` (properties at the current `t`, novelty included, optionally grouped by other properties) and, under the `iceberg` feature, `Fluree::profile_table` (lake columns streamed through the graph-source scan, pinned to the current snapshot). Both report the same `ProfileReport` shape.
…e text values The t-digest kept its extremes as ±infinity, which serde_json writes as null and refuses to read back into an f64, so any column profile with no numeric values (a text or ref column) could not be deserialised. The extremes are now options, like the moments already were. A bytes cell was folded to one placeholder, so a bytes column profiled as a single constant value. Bytes now hash on their content under their own kind and report a hex sample. The text extremes compared incoming values against truncated samples, which mis-ordered strings that share a prefix longer than the sample. The whole value is kept for the comparison and truncated only in the summary. Text lengths count characters rather than bytes. `ProfileValue::to_text` replaces the three copies of the display match. The stats crate takes xxhash-rust from the workspace. On the API side, group keys are interned so the subject-to-key map costs one pointer per subject; the module doc states that the ledger face holds one property's flakes in memory per pass and that neither face applies view policy. Table profiling reads the schema first, so an empty table still reports missing columns and rejects an unknown group-by column, and the accumulators no longer depend on a first batch arriving.
aaj3f
left a comment
aaj3f
left a comment
There was a problem hiding this comment.
@bplatz as a design idea to move this to its own crate & kernel seems very good to me and I have no pushback there. My only investigations were into memory performance (sizing) against that kernel. With Claude's assistance I found what I think may be some significant items worth addressing before merge:
This is a genuinely good kernel: const-generic Hll<M> with the indexer's Hll256 as an instantiation, exact Welford/Chan moments, a merging t-digest with the k_1 scale, Misra–Gries with the Agarwal merge and an honest decrements error bar, and every summary merges — which is what makes shard-parallel profiling and incremental refresh possible later. The thing I checked hardest was the persisted-sketch claim, and it holds: the deleted hll256.rs copied verbatim beside the new Hll256 and driven with the same 1 M hashes is byte-identical on registers, estimates, merges, every estimator branch and every error path in both directions, the indexer's hash lives in hashing.rs (untouched), and the per-flake insert is the same #[inline] code as before. Commit 2's fixes (Option extremes, whole-value text comparison, bytes on content, character lengths, schema-first table profiling) all check out under the crate's own tests, and no server route or tool reaches the two faces yet, so the "no view policy" statement is not a bypass — just a constraint for whoever wires them.
The two I have to block on: findings::scale_drift panics on a group with a negative median — ratio.ln() is NaN, partial_cmp falls back to Equal, and std's sort panics on the non-transitive comparator (reproduced on 1.97 with 200 groups, every third negative; a caller cannot guard it from outside the library) — and a ColumnProfile is sized for a column, not a group: an inline Hll4096, a HeavyHitters map that allocates ~5 KB eagerly in new, and t-digest capacity that is never shrunk come to 11–30 KB per group measured, times a 100,000-group default per column, so the per-(part, division) baseline the module doc advertises is 1.1–2.9 GiB per profiled column. Each fix is a few lines (skip median <= 0 and sort with total_cmp; HashMap::new(), shrink_to_fit, a smaller or lazy per-group HLL, a 10k default).
Fold-ins alongside, all inline: the lake face rebuilds the group key per row per profiled column with an allocation per key cell; profile_table loads the table three times so snapshot_id isn't structurally the snapshot scanned; the single-key subject map is deep-cloned while its source is alive; sum is the one bare float left that overflows to inf and breaks the round-trip; and a max_values guard on profile_ledger until a streaming walk exists (that walk needs a core API — genuinely separate scope, and I'd name it as such). Optional: SipHash over already-hashed keys, seven quantile() compress-clones per summary, an unclamped compression on deserialise, and one indexer-side monotonicity assert so the indexer's own tests can see a broken sketch merge.
Adherence to repo commitments:
- Patterns/abstractions: ✔ one kernel every column summary shares; the indexer re-exports rather than duplicates;
[lints] workspace = true, workspace version/license,tabularas an optional feature, zerounsafe. - Performance (speed first, memory second): ✔ indexer per-flake path unchanged (no bench budget needed); ✖ CRITICAL on memory for grouped profiling as shipped (per-group cost × default cap);
⚠️ N × K key allocations per row on the lake face, whole-property materialisation on the ledger face. - Testing: ✔ 47 kernel tests, 4 ledger-face and 2 lake-face integration tests wired into run targets (
grp_ledger, the[[test]]withrequired-features), CI green; ✖ the negative-median panic and the memory cliff are untested;⚠️ the indexer's tests don't detect a broken HLL merge. - Conventions: ✔ two thorough commits; module docs state the memory model and the no-policy caveat;
⚠️ struct docs /#[must_use]thin on the kernel's public API.
Verified locally at branch HEAD 71727bf69: compatibility proof by scratch test against the base commit's file (deleted after); cargo test -p fluree-db-stats 47, -p fluree-db-indexer stats 24 + hll 1, -p fluree-db-api --features iceberg,native --test grp_ledger profile 4 (+1 scratch novelty-retraction test), --test it_iceberg_local_fs 2; cargo fmt --all -- --check and cargo clippy -p fluree-db-stats -p fluree-db-indexer --all-targets --no-deps -- -D warnings clean; the panic, the memory numbers (counting allocator) and the sum round-trip failure reproduced; HLL merge mutation restored; worktree clean.
Just be sure to get the sort guard and the per-group sizing in before this merges — #1791 and #1792 rebase underneath it — and I'd rather see the fold-ins here than in the backlog.
| }) | ||
| }) | ||
| .collect(); | ||
| out.sort_by(|a, b| { |
There was a problem hiding this comment.
Blocking — scale_drift panics on a group with a negative median. ratio = median / overall (:201) is guarded only for overall <= 0 (:189). A group whose median is negative gives ratio < 0, ln = NaN, partial_cmp → None → Equal, and with NaN and non-NaN elements mixed the comparator is non-transitive, so std's sort (≥ 1.81) panics: "user-provided comparison function does not correctly implement a total order" (core/src/slice/sort/shared/smallsort.rs).
Reproduced on rustc 1.97: 200 groups × 5 values, every third group at -10-i, the rest at 10+i (overall median 59.5) → panic instead of a sorted Vec<ScaleDrift>. Any column of deltas, balances, adjustments or temperatures grouped by anything hits it, and the caller (a quality rule) cannot guard it from outside the library.
Fix: skip groups with median <= 0.0 — log-ratio drift is undefined there, which is what the overall guard already encodes — and sort on a precomputed |ln ratio| with f64::total_cmp so a NaN can never reach sort_by. Add the negative-median group as a test.
| config: ProfileConfig, | ||
| count: u64, | ||
| kinds: [u64; 9], | ||
| distinct: Hll4096, |
There was a problem hiding this comment.
CRITICAL (memory) — a ColumnProfile is sized for a column, not a group, and the grouped default makes 100,000 of them per column. Measured with a counting global allocator: 100,000 groups × 1 value = 1,081 MiB (11.3 KB per group); 10,000 groups × 600 values = 288 MiB (30 KB per group); size_of::<ColumnProfile>() = 4,536 B.
Where it goes: this inline Hll4096 (4,096 B in every profile, including a one-value group); HeavyHitters::new's HashMap::with_capacity(capacity + 1) (heavy_hitters.rs:57), which hashbrown rounds to 128 slots ≈ 5 KB allocated eagerly before the first observe; and the t-digest's Vec::with_capacity(points.len()) (tdigest.rs:116) sized for ~600 inputs but holding ~100 centroids, plus the emptied 500-slot buffer keeping its allocation (:112) — a digest that has seen ≥ 500 values holds ≈ 18 KB. DEFAULT_MAX_GROUPS = 100_000 (grouped.rs:18) is the API default (fluree-db-api/src/profile.rs:62), and profile_ledger builds one accumulator per requested column — so the "per-(part, division) baseline" the module doc advertises, over 10 properties and ≥ 100k parts, allocates 10 × 1.1–2.9 GiB before overflow starts pooling. GroupedProfile::merge (grouped.rs:107) clones whole Groups as well.
Four independent, cheap fixes: HashMap::new() in HeavyHitters::new (a one-value group then costs ~100 B); shrink_to_fit after compress and drop the emptied buffer; a smaller or lazily allocated per-group HLL (Hll256 / Hll1024, or Option/Box allocated on first insert) with Hll4096 kept for total; and a 10,000 default with the per-group cost documented beside it so max_groups is the knob callers reach for.
| .iter() | ||
| .filter_map(|k| batch.column_by_name(k)) | ||
| .collect(); | ||
| for (name, acc) in names.iter().zip(accs.iter_mut()) { |
There was a problem hiding this comment.
Should-fix (performance, lake face). The group key is rebuilt inside the per-column loop — once per row per profiled column — and tabular::display_at (tabular.rs:44-46, value_at(col,row).to_text()) allocates a String per key cell even for a Column::String where a borrow would do. That is N profiled columns × K key columns allocations per row for keys that are identical across the N columns: 20 columns, one key, 100 M rows → 2 × 10⁹ allocations. profile_column_grouped in tabular.rs:57-69 has the same shape.
Build the batch's keys once (a Vec<String> per batch, or one arena with offsets) and run the column loop over &keys[row]; let to_text return Cow<str> so a string cell borrows.
| use futures::StreamExt; | ||
|
|
||
| let provider = crate::graph_source::FlureeR2rmlProvider::new(self); | ||
| let snapshot_id = provider |
There was a problem hiding this comment.
Should-fix. profile_table resolves the table three times: current_snapshot_id and table_column_names each go through prepare_iceberg_scan (graph_source/r2rml.rs:933: nameservice lookup + catalog loadTable), then scan_table (:419-421) goes through the session-cached load_table_context. A commit landing between the calls makes the report say snapshot X and profile snapshot Y — the doc at :349-351 promises "at the table's current snapshot" — and each call pays two extra REST loadTables (priced at ~1–3 s each in the comment at r2rml.rs:2430).
Resolve metadata once (the session-pinned load_table_context), read current_snapshot() and the schema from that TableMetadata, and drive the scan from the same pinned context. Also: when every requested column is unknown the projection is empty (:403-411) and the scan still reads the whole table for nothing — short-circuit when no accumulator exists.
| } | ||
| maps.push(map); | ||
| } | ||
| let keys: Option<HashMap<Sid, Arc<str>>> = match maps.as_slice() { |
There was a problem hiding this comment.
Should-fix (memory, ledger face). [single] => Some(single.clone()) deep-clones the whole HashMap<Sid, Arc<str>> while maps is still alive — 2× peak for the common one-key case — and each entry is a cloned Sid (24 B) plus a 16 B Arc<str> plus hashbrown overhead, ≈ 56–64 B per subject rather than the "one pointer per subject" the commit says (the interning does make the string one-per-distinct-key, which is the useful part). maps.into_iter().next() / std::mem::take for the single case; build the joined map by draining first.
Related and larger: property_flakes (:207-224) materialises every current assertion of the property as Vec<Flake> with no cap — the module doc says so, and no streaming range API exists yet, so it's a stated scope decision — but a max_values on ProfileRequest that probes with RangeOptions::with_flake_limit(limit + 1) and refuses would turn an OOM into an error today, with the streaming walk as the tracked follow-up (naming it: needs a core API, genuinely separate scope).
| pub max: f64, | ||
| pub mean: f64, | ||
| pub stddev: Option<f64>, | ||
| pub sum: f64, |
There was a problem hiding this comment.
Should-fix (serde edge). sum is the one bare float left that can be non-finite: two Float(1e308) observations overflow it to inf, serde_json writes "sum":null, and from_str::<ColumnSummary> then fails — the report no longer round-trips, which is exactly what commit 2 fixed for the extremes. mean is Welford-safe and min/max are finite. Make it Option<f64> (None when non-finite), or serialise with a helper that nulls non-finite AND reads back as Option.
| /// that formats the value lazily. | ||
| #[inline] | ||
| pub fn observe(&mut self, hash: u64, sample: impl FnOnce() -> String) { | ||
| if let Some(c) = self.counters.get_mut(&hash) { |
There was a problem hiding this comment.
Optional (perf on the profiling path). HashMap<u64, Counter> with the default hasher re-hashes an xxh3 key on every observe, merge and count_of; a ten-line NoHash BuildHasher removes ~15–20 ns per value. And when the table is full and the value is unseen, the retain at :81-86 runs over every counter per value — Misra–Gries' known cost, amortised O(1) for a mostly-unique column but O(capacity) per singleton when ~64 heavy values persist; worth a comment. Also a doc line on top(): empty under !is_exact() means "nothing provably frequent" — the equal-shard merge correctly drops every counter (decrements = 10, bound holds), which reads like a bug until you know.
| if self.total == 0.0 { | ||
| return None; | ||
| } | ||
| if !self.buffer.is_empty() { |
There was a problem hiding this comment.
Optional (perf). quantile() clones and compresses the digest (a sort of ≤ 600 centroids) on every call while the buffer is non-empty, and summary() (profile.rs:305-311) calls it seven times per column per group; findings::baseline_of and scale_drift repeat it per group. Compress once in summary (clone once, read all seven) or expose quantiles(&[f64]). Related: compression is clamped to ≥ 10 only in new (:48-50) — a digest deserialised with "compression":0.0 never merges (q_limit is NaN; 5,000 points → 5,000 centroids); a deserialize_with clamp closes it.
| self.count += other.count; | ||
| self.values_hll.merge_inplace(&other.values_hll); | ||
| self.subjects_hll.merge_inplace(&other.subjects_hll); | ||
| self.values_hll.merge(&other.values_hll); |
There was a problem hiding this comment.
Praise, and the thing I checked hardest. The compatibility claim is true, and it was proven rather than read: the deleted hll256.rs was copied verbatim from the base commit into a scratch test beside the new Hll256 and both were driven with the same 1 M hashes — byte-identical registers, identical estimates and merges, identical behaviour on every estimator branch and every from_bytes_versioned error path, in both directions. The indexer never hashed inside that file (insert_hash takes the u64 from stats/hashing.rs, untouched), and this merge is the only edit on the refresh path; the per-flake insert_hash at :447-448 is the same #[inline] fixed-array code as before. One coverage note: flipping the kernel's merge max to min reddens two kernel tests and none of the 390 indexer tests — a single estimate-monotonicity assert on the merge_from test would let the indexer see its own sketch break.
| /// Hash a value under its kind's domain. Nulls have no hash; callers | ||
| /// count them separately and never call this for them. | ||
| #[inline] | ||
| pub fn value_hash(value: &ProfileValue<'_>) -> Option<u64> { |
There was a problem hiding this comment.
Praise. Domain-separated hashing with integral-float folding (7 == 7.0 == 7.00, -0.0 == 0.0, "7" ≠ 7 ≠ <7>) is the right definition of "the same value" for a profile, and it is entirely separate from the indexer's persisted-sketch hash, which is why the compatibility question was moot.
2 participants
Summary
Adds a
fluree-db-statscrate holding the mergeable sketches every column summary in Fluree can share, and two on-demand profiling entry points influree-db-apibuilt on it.The kernel (
fluree-db-stats):Hll<M>: const-generic HyperLogLog.Hll256is the indexer's existing per-(graph, property) sketch,Hll4096the profiling default.Moments: exact count / mean / variance / min / max / sum (Welford single-stream, Chan merge).TDigest: merging t-digest with thek_1scale function; tight at the tails.HeavyHitters: Misra–Gries frequent values with the Agarwal et al. merge; exact when distinct ≤ capacity, bounded undercount otherwise, and every answer carries its own error bar.ColumnProfilecomposes the four over one column;GroupedProfilekeeps one per group key with a bounded group count and an overflow pool.findings: group baselines, ratio and z-score rules, concentration, per-group constancy, scale drift.tabularfeature: a face overfluree-db-tabularcolumn batches.Every summary merges: two profiles over disjoint shards fold into the profile of the union, exactly for counts and moments, within stated error for the sketches.
The indexer now re-exports the kernel's
Hll256asHllSketch256. Register layout and the 256-byte persisted form are unchanged, so existing sketches load and merge as before; the only call-site change wasmerge_inplace→merge.The API gains:
Fluree::profile_ledger: properties at the currentt, novelty included, optionally grouped by one or more other properties.Fluree::profile_table(featureiceberg): lake columns streamed through the graph-source scan, pinned to the current snapshot, optionally grouped by other columns.Both return the same
ProfileReportshape.Limitations stated in the docs
Test plan
cargo fmt --all --checkcargo clippy -p fluree-db-stats -p fluree-db-indexer -p fluree-db-api --features fluree-db-api/iceberg --all-targets --no-depscargo test -p fluree-db-stats --all-features(51 tests: sketch error bounds at both HLL sizes, merge-equals-whole for every sketch, JSON round trips including empty digests and text-only profiles, whole-value text extremes, bytes by content)cargo test -p fluree-db-api --test grp_ledger it_profile_ledger(flat, grouped, two-key grouping with constancy, unknown group property)cargo test -p fluree-db-api --features iceberg --test it_iceberg_local_fs local_table_profiles(all columns, grouped with a missing column)