mz_persist_client/
stats.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Aggregate statistics about data stored in persist.
11
12use std::borrow::Cow;
13use std::sync::Arc;
14
15use mz_dyncfg::{Config, ConfigSet, ParameterScope};
16
17use crate::batch::UntrimmableColumns;
18use crate::metrics::Metrics;
19use crate::read::LazyPartStats;
20
21use crate::ShardId;
22
23/// Percent of filtered data to opt in to correctness auditing.
24pub(crate) const STATS_AUDIT_PERCENT: Config<usize> = Config::new(
25    "persist_stats_audit_percent",
26    1,
27    "Percent of filtered data to opt in to correctness auditing (Materialize).",
28    ParameterScope::Environment,
29);
30
31/// See description for usage.
32pub const STATS_AUDIT_PANIC: Config<bool> = Config::new(
33    "persist_stats_audit_panic",
34    true,
35    "If set (as it is by default), panic on any auditing failure. If not, report an error but \
36    pass along the data as normal. This should almost certainly be paired with an audit rate of 100%, \
37    so all parts are audited, for consistency.",
38    ParameterScope::Environment,
39);
40
41/// Computes and stores statistics about each batch part.
42///
43/// These can be used at read time to entirely skip fetching a part based on its
44/// statistics. See [STATS_FILTER_ENABLED].
45pub(crate) const STATS_COLLECTION_ENABLED: Config<bool> = Config::new(
46    "persist_stats_collection_enabled",
47    true,
48    "\
49    Whether to calculate and record statistics about the data stored in \
50    persist to be used at read time, see persist_stats_filter_enabled \
51    (Materialize).",
52    ParameterScope::Environment,
53);
54
55/// Uses previously computed statistics about batch parts to entirely skip
56/// fetching them at read time.
57///
58/// See `STATS_COLLECTION_ENABLED`.
59pub const STATS_FILTER_ENABLED: Config<bool> = Config::new(
60    "persist_stats_filter_enabled",
61    true,
62    "\
63    Whether to use recorded statistics about the data stored in persist to \
64    filter at read time, see persist_stats_collection_enabled (Materialize).",
65    ParameterScope::Environment,
66);
67
68/// The budget (in bytes) of how many stats to write down per batch part. When
69/// the budget is exceeded, stats will be trimmed away according to a variety of
70/// heuristics.
71pub(crate) const STATS_BUDGET_BYTES: Config<usize> = Config::new(
72    "persist_stats_budget_bytes",
73    1024,
74    "The budget (in bytes) of how many stats to maintain per batch part.",
75    ParameterScope::Environment,
76);
77
78pub(crate) const STATS_UNTRIMMABLE_COLUMNS_EQUALS: Config<fn() -> String> = Config::new(
79    "persist_stats_untrimmable_columns_equals",
80    || {
81        [
82            // If we trim the "err" column, then we can't ever use pushdown on a
83            // part (because it could have >0 errors).
84            "err",
85            "ts",
86            "receivedat",
87            "createdat",
88            // Fivetran created tables track deleted rows by setting this column.
89            //
90            // See <https://fivetran.com/docs/using-fivetran/features#capturedeletes>.
91            "_fivetran_deleted",
92        ]
93        .join(",")
94    },
95    "\
96    Which columns to always retain during persist stats trimming. Any column \
97    with a name exactly equal (case-insensitive) to one of these will be kept. \
98    Comma separated list.",
99    ParameterScope::Environment,
100);
101
102pub(crate) const STATS_UNTRIMMABLE_COLUMNS_PREFIX: Config<fn() -> String> = Config::new(
103    "persist_stats_untrimmable_columns_prefix",
104    || ["last_"].join(","),
105    "\
106    Which columns to always retain during persist stats trimming. Any column \
107    with a name starting with (case-insensitive) one of these will be kept. \
108    Comma separated list.",
109    ParameterScope::Environment,
110);
111
112pub(crate) const STATS_UNTRIMMABLE_COLUMNS_SUFFIX: Config<fn() -> String> = Config::new(
113    "persist_stats_untrimmable_columns_suffix",
114    || ["timestamp", "time", "_at", "_tstamp"].join(","),
115    "\
116    Which columns to always retain during persist stats trimming. Any column \
117    with a name ending with (case-insensitive) one of these will be kept. \
118    Comma separated list.",
119    ParameterScope::Environment,
120);
121
122pub(crate) fn untrimmable_columns(cfg: &ConfigSet) -> UntrimmableColumns {
123    fn split(x: String) -> Vec<Cow<'static, str>> {
124        x.split(',')
125            .filter(|x| !x.is_empty())
126            .map(|x| x.to_owned().into())
127            .collect()
128    }
129    UntrimmableColumns {
130        equals: split(STATS_UNTRIMMABLE_COLUMNS_EQUALS.get(cfg)),
131        prefixes: split(STATS_UNTRIMMABLE_COLUMNS_PREFIX.get(cfg)),
132        suffixes: split(STATS_UNTRIMMABLE_COLUMNS_SUFFIX.get(cfg)),
133    }
134}
135
136/// Statistics about the contents of a shard as_of some time.
137///
138/// TODO: Add more stats here as they become necessary.
139#[derive(Debug)]
140pub struct SnapshotStats {
141    /// The shard these statistics are for.
142    pub shard_id: ShardId,
143    /// An estimate of the count of updates in the shard.
144    ///
145    /// This is an upper bound on the number of updates that persist_source
146    /// would emit if you snapshot the source at the given as_of. The real
147    /// number of updates, after consolidation, might be lower. It includes both
148    /// additions and retractions.
149    ///
150    /// NB: Because of internal persist compaction, the answer for a given as_of
151    /// may change over time (as persist advances through Seqnos), but because
152    /// compaction never results in more updates than the sum of the inputs, it
153    /// can only go down.
154    pub num_updates: usize,
155}
156
157/// Statistics about the contents of the parts of a shard as_of some time.
158#[derive(Debug)]
159pub struct SnapshotPartsStats {
160    /// Metrics for the persist backing shard, so the caller can report any
161    /// necessary counters.
162    pub metrics: Arc<Metrics>,
163    /// The shard these statistics are for.
164    pub shard_id: ShardId,
165    /// Stats for individual parts.
166    pub parts: Vec<SnapshotPartStats>,
167}
168
169/// Part-specific stats.
170#[derive(Debug)]
171pub struct SnapshotPartStats {
172    /// The size of the encoded data in bytes.
173    pub encoded_size_bytes: usize,
174    /// The raw/encoded statistics for that part, if we have them.
175    pub stats: Option<LazyPartStats>,
176}
mz_persist_client/stats.rs

mz_persist_client/
stats.rs
