faucet_core/quality/config.rs
1//! Config-shaped types for the data-quality layer. Pure declarations — no
2//! evaluation logic (that lives in `record.rs` / `batch.rs`) and no
3//! compilation (that lives in `compile.rs`).
4
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9/// What to do when a check fails. The allowed subset is validated per check
10/// at compile time (see `compile.rs`).
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
12#[serde(rename_all = "snake_case")]
13pub enum OnFailure {
14 /// Route the specific offending row(s) to the DLQ; keep the rest.
15 Quarantine,
16 /// Route all survivors of the page to the DLQ; write nothing this page.
17 QuarantineBatch,
18 /// Surface `FaucetError::QualityFailure` and fail the run.
19 Abort,
20}
21
22/// Ordering / equality operator for the `compare` check.
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
24#[serde(rename_all = "snake_case")]
25pub enum CompareOp {
26 /// Greater than: `field > value`. Both must be JSON numbers.
27 Gt,
28 /// Greater than or equal: `field >= value`. Both must be JSON numbers.
29 Gte,
30 /// Less than: `field < value`. Both must be JSON numbers.
31 Lt,
32 /// Less than or equal: `field <= value`. Both must be JSON numbers.
33 Lte,
34 /// JSON equality. Two numbers compare by numeric value (`1` == `1.0`, and
35 /// large 64-bit integers compare exactly); all other types compare
36 /// structurally with no cross-type coercion (string `"5"` != number `5`).
37 Eq,
38 /// JSON inequality — the negation of [`CompareOp::Eq`] (numbers by value,
39 /// other types structurally).
40 Ne,
41}
42
43impl std::fmt::Display for CompareOp {
44 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45 f.write_str(match self {
46 CompareOp::Gt => "gt",
47 CompareOp::Gte => "gte",
48 CompareOp::Lt => "lt",
49 CompareOp::Lte => "lte",
50 CompareOp::Eq => "eq",
51 CompareOp::Ne => "ne",
52 })
53 }
54}
55
56/// Expected JSON type for the `type_is` check.
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
58#[serde(rename_all = "snake_case")]
59pub enum JsonType {
60 /// JSON boolean (`true` / `false`).
61 Boolean,
62 /// JSON number (integer or float).
63 Number,
64 /// JSON string.
65 String,
66 /// JSON array.
67 Array,
68 /// JSON object.
69 Object,
70 /// JSON null. Note: a *missing* field is distinct from an explicit `null`.
71 Null,
72}
73
74impl std::fmt::Display for JsonType {
75 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
76 f.write_str(match self {
77 JsonType::Boolean => "boolean",
78 JsonType::Number => "number",
79 JsonType::String => "string",
80 JsonType::Array => "array",
81 JsonType::Object => "object",
82 JsonType::Null => "null",
83 })
84 }
85}
86
87fn default_true() -> bool {
88 true
89}
90
91/// The `quality:` config block. Per-record checks run first (partitioning the
92/// page into survivors + quarantined); per-batch checks then run over the
93/// survivors.
94#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
95#[serde(deny_unknown_fields)]
96pub struct QualitySpec {
97 /// Per-record checks, evaluated in declared order (first failure wins).
98 #[serde(default)]
99 pub record: Vec<RecordCheck>,
100 /// Per-batch checks, evaluated per page over the survivors.
101 #[serde(default)]
102 pub batch: Vec<BatchCheck>,
103}
104
105/// A per-record check. Addressed field accepts the filter/explode path subset
106/// (bare key, `dot.path`, `$['bracketed']`).
107///
108/// Every variant carries `field` (the path to check) and `on_failure` (what a
109/// failure does). Per-record checks accept only [`OnFailure::Quarantine`] or
110/// [`OnFailure::Abort`] — `quarantine_batch` is rejected at compile time
111/// because a per-record failure is always attributable to one row.
112#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
113#[serde(tag = "type", rename_all = "snake_case")]
114#[serde(deny_unknown_fields)]
115pub enum RecordCheck {
116 /// Field present and non-null.
117 NotNull {
118 /// Path to the checked field.
119 field: String,
120 /// When `true` (default) a missing field fails; when `false` only an
121 /// explicit JSON `null` fails.
122 #[serde(default = "default_true")]
123 treat_missing_as_null: bool,
124 /// What a failure does: `quarantine` (row to the DLQ) or `abort`.
125 on_failure: OnFailure,
126 },
127 /// Field is a string, non-empty after `trim()`.
128 NotEmpty {
129 /// Path to the checked field. A missing field, a `null`, or a
130 /// non-string value all fail.
131 field: String,
132 /// What a failure does: `quarantine` (row to the DLQ) or `abort`.
133 on_failure: OnFailure,
134 },
135 /// Field is a string matching `pattern`.
136 RegexMatch {
137 /// Path to the checked field. A missing field or a non-string value
138 /// fails.
139 field: String,
140 /// Rust `regex`-crate pattern. Unanchored — it only has to match
141 /// *somewhere* in the value, so anchor with `^…$` for a full match.
142 /// An invalid pattern is rejected at config load, never mid-run.
143 pattern: String,
144 /// What a failure does: `quarantine` (row to the DLQ) or `abort`.
145 on_failure: OnFailure,
146 },
147 /// Field value is a member of `values` (exact JSON equality).
148 ValueInSet {
149 /// Path to the checked field. A missing field fails.
150 field: String,
151 /// The allowed values. Must be non-empty. Compared by JSON equality
152 /// with no type coercion, so the string `"5"` does not match the
153 /// number `5`.
154 values: Vec<Value>,
155 /// What a failure does: `quarantine` (row to the DLQ) or `abort`.
156 on_failure: OnFailure,
157 },
158 /// Field value is NOT a member of `values` (exact JSON equality).
159 NotInSet {
160 /// Path to the checked field. A **missing** field passes — there is no
161 /// value to forbid.
162 field: String,
163 /// The forbidden values. Must be non-empty. Compared by JSON equality
164 /// with no type coercion.
165 values: Vec<Value>,
166 /// What a failure does: `quarantine` (row to the DLQ) or `abort`.
167 on_failure: OnFailure,
168 },
169 /// Field value compares against `value` under `op`.
170 Compare {
171 /// Path to the checked field. A missing field fails.
172 field: String,
173 /// The comparison to apply, as `field <op> value`.
174 op: CompareOp,
175 /// The right-hand side of the comparison. The ordering ops
176 /// (`gt`/`gte`/`lt`/`lte`) require a JSON number here — anything else
177 /// is rejected at config load — and also require the record's value to
178 /// be a number; `eq`/`ne` accept any JSON value.
179 value: Value,
180 /// What a failure does: `quarantine` (row to the DLQ) or `abort`.
181 on_failure: OnFailure,
182 },
183 /// Field's JSON type equals `expected`.
184 TypeIs {
185 /// Path to the checked field. A missing field fails, which is distinct
186 /// from an `expected: null` match on a present `null`.
187 field: String,
188 /// The required JSON type.
189 expected: JsonType,
190 /// What a failure does: `quarantine` (row to the DLQ) or `abort`.
191 on_failure: OnFailure,
192 },
193 /// Field is a string whose char count is within `[min, max]`.
194 StringLength {
195 /// Path to the checked field. A missing field or a non-string value
196 /// fails.
197 field: String,
198 /// Inclusive minimum length in Unicode **characters** (not bytes).
199 /// Omit for no lower bound.
200 #[serde(default)]
201 min: Option<usize>,
202 /// Inclusive maximum length in Unicode **characters** (not bytes).
203 /// Omit for no upper bound. At least one of `min`/`max` is required,
204 /// and `min <= max`, both enforced at config load.
205 #[serde(default)]
206 max: Option<usize>,
207 /// What a failure does: `quarantine` (row to the DLQ) or `abort`.
208 on_failure: OnFailure,
209 },
210 /// The whole record validates against a JSON Schema document.
211 #[cfg(feature = "quality-jsonschema")]
212 JsonSchema {
213 /// The JSON Schema document, inline. Compiled once at config load —
214 /// an invalid schema fails there, not on the first page. The first
215 /// validation error becomes the DLQ/abort message.
216 schema: Value,
217 /// What a failure does: `quarantine` (row to the DLQ) or `abort`.
218 on_failure: OnFailure,
219 },
220}
221
222/// A per-batch check, evaluated per page over the survivors of the per-record
223/// pass.
224///
225/// **Scope is one page, not the whole run** — with a source `batch_size` of
226/// 1000, `row_count` sees 1000 rows at a time. Set the source's
227/// `batch_size: 0` to evaluate these over the entire result set instead.
228///
229/// The aggregate checks (`row_count` / `null_rate` / `distinct_count`) cannot
230/// blame an individual row, so they accept only [`OnFailure::Abort`] or
231/// [`OnFailure::QuarantineBatch`]; `unique` *is* row-attributable and so
232/// accepts `quarantine` or `abort`. Either way the wrong choice is rejected at
233/// config load.
234#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
235#[serde(tag = "type", rename_all = "snake_case")]
236#[serde(deny_unknown_fields)]
237pub enum BatchCheck {
238 /// Survivor count is within `[min, max]` (at least one bound required).
239 RowCount {
240 /// Inclusive minimum surviving rows in the page. Omit for no lower
241 /// bound.
242 #[serde(default)]
243 min: Option<usize>,
244 /// Inclusive maximum surviving rows in the page. Omit for no upper
245 /// bound. At least one of `min`/`max` is required, and `min <= max`,
246 /// both enforced at config load.
247 #[serde(default)]
248 max: Option<usize>,
249 /// What a failure does: `abort`, or `quarantine_batch` (every survivor
250 /// in the page goes to the DLQ and nothing is written).
251 on_failure: OnFailure,
252 },
253 /// Null-or-missing rate of `field` across survivors is `<= max`.
254 NullRate {
255 /// Path to the measured field. A missing field counts the same as an
256 /// explicit `null`.
257 field: String,
258 /// Maximum allowed null-or-missing proportion, in `[0.0, 1.0]`. Out-of-range values are rejected at compile time.
259 max: f64,
260 /// What a failure does: `abort`, or `quarantine_batch` (every survivor
261 /// in the page goes to the DLQ and nothing is written).
262 on_failure: OnFailure,
263 },
264 /// The composite `fields` tuple is unique across survivors.
265 Unique {
266 /// Paths forming the uniqueness key, in order. One path for a simple
267 /// key, several for a composite one. Must be non-empty. A missing
268 /// field is a distinct key value from an explicit `null`.
269 fields: Vec<String>,
270 /// What a failure does. This check names the offending rows, so
271 /// `quarantine` sends the **duplicate occurrences** to the DLQ (the
272 /// first occurrence of each key is kept); `abort` fails the run.
273 on_failure: OnFailure,
274 },
275 /// Distinct values of `field` across survivors is within `[min, max]`.
276 DistinctCount {
277 /// Path to the counted field. A missing field counts as its own
278 /// distinct value, separate from an explicit `null`.
279 field: String,
280 /// Inclusive minimum number of distinct values. Omit for no lower
281 /// bound.
282 #[serde(default)]
283 min: Option<usize>,
284 /// Inclusive maximum number of distinct values. Omit for no upper
285 /// bound. At least one of `min`/`max` is required, and `min <= max`,
286 /// both enforced at config load.
287 #[serde(default)]
288 max: Option<usize>,
289 /// What a failure does: `abort`, or `quarantine_batch` (every survivor
290 /// in the page goes to the DLQ and nothing is written).
291 on_failure: OnFailure,
292 },
293}
294
295#[cfg(test)]
296mod tests {
297 use super::*;
298
299 #[test]
300 fn on_failure_serializes_snake_case() {
301 assert_eq!(
302 serde_json::to_string(&OnFailure::QuarantineBatch).unwrap(),
303 "\"quarantine_batch\""
304 );
305 }
306
307 #[test]
308 fn compare_op_round_trips() {
309 let op: CompareOp = serde_json::from_str("\"gte\"").unwrap();
310 assert_eq!(op, CompareOp::Gte);
311 }
312
313 #[test]
314 fn json_type_round_trips() {
315 let t: JsonType = serde_json::from_str("\"boolean\"").unwrap();
316 assert_eq!(t, JsonType::Boolean);
317 }
318
319 #[test]
320 fn parses_full_quality_block() {
321 let spec: QualitySpec = serde_json::from_value(serde_json::json!({
322 "record": [
323 { "type": "not_null", "field": "user_id", "on_failure": "quarantine" },
324 { "type": "compare", "field": "age", "op": "gte", "value": 0, "on_failure": "abort" },
325 { "type": "string_length", "field": "name", "min": 1, "max": 256, "on_failure": "quarantine" }
326 ],
327 "batch": [
328 { "type": "row_count", "min": 1, "max": 100000, "on_failure": "abort" },
329 { "type": "unique", "fields": ["id"], "on_failure": "quarantine" }
330 ]
331 }))
332 .unwrap();
333 assert_eq!(spec.record.len(), 3);
334 assert_eq!(spec.batch.len(), 2);
335 assert!(matches!(spec.record[0], RecordCheck::NotNull { .. }));
336 assert!(matches!(spec.batch[1], BatchCheck::Unique { .. }));
337 if let RecordCheck::NotNull {
338 treat_missing_as_null,
339 ..
340 } = &spec.record[0]
341 {
342 assert!(
343 *treat_missing_as_null,
344 "treat_missing_as_null defaults to true"
345 );
346 } else {
347 panic!("expected first record check to be NotNull");
348 }
349 }
350
351 #[test]
352 fn empty_quality_block_defaults_to_no_checks() {
353 let spec: QualitySpec = serde_json::from_str("{}").unwrap();
354 assert!(spec.record.is_empty());
355 assert!(spec.batch.is_empty());
356 }
357}