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)]
95pub struct QualitySpec {
96    /// Per-record checks, evaluated in declared order (first failure wins).
97    #[serde(default)]
98    pub record: Vec<RecordCheck>,
99    /// Per-batch checks, evaluated per page over the survivors.
100    #[serde(default)]
101    pub batch: Vec<BatchCheck>,
102}
103
104/// A per-record check. Addressed field accepts the filter/explode path subset
105/// (bare key, `dot.path`, `$['bracketed']`).
106#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
107#[serde(tag = "type", rename_all = "snake_case")]
108pub enum RecordCheck {
109    /// Field present and non-null.
110    NotNull {
111        field: String,
112        /// When `true` (default) a missing field fails; when `false` only an
113        /// explicit JSON `null` fails.
114        #[serde(default = "default_true")]
115        treat_missing_as_null: bool,
116        on_failure: OnFailure,
117    },
118    /// Field is a string, non-empty after `trim()`.
119    NotEmpty {
120        field: String,
121        on_failure: OnFailure,
122    },
123    /// Field is a string matching `pattern`.
124    RegexMatch {
125        field: String,
126        pattern: String,
127        on_failure: OnFailure,
128    },
129    /// Field value is a member of `values` (exact JSON equality).
130    ValueInSet {
131        field: String,
132        values: Vec<Value>,
133        on_failure: OnFailure,
134    },
135    /// Field value is NOT a member of `values` (exact JSON equality).
136    NotInSet {
137        field: String,
138        values: Vec<Value>,
139        on_failure: OnFailure,
140    },
141    /// Field value compares against `value` under `op`.
142    Compare {
143        field: String,
144        op: CompareOp,
145        value: Value,
146        on_failure: OnFailure,
147    },
148    /// Field's JSON type equals `expected`.
149    TypeIs {
150        field: String,
151        expected: JsonType,
152        on_failure: OnFailure,
153    },
154    /// Field is a string whose char count is within `[min, max]`.
155    StringLength {
156        field: String,
157        #[serde(default)]
158        min: Option<usize>,
159        #[serde(default)]
160        max: Option<usize>,
161        on_failure: OnFailure,
162    },
163    /// The whole record validates against a JSON Schema document.
164    #[cfg(feature = "quality-jsonschema")]
165    JsonSchema {
166        schema: Value,
167        on_failure: OnFailure,
168    },
169}
170
171/// A per-batch check, evaluated per page over the survivors of the per-record
172/// pass.
173#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
174#[serde(tag = "type", rename_all = "snake_case")]
175pub enum BatchCheck {
176    /// Survivor count is within `[min, max]` (at least one bound required).
177    RowCount {
178        #[serde(default)]
179        min: Option<usize>,
180        #[serde(default)]
181        max: Option<usize>,
182        on_failure: OnFailure,
183    },
184    /// Null-or-missing rate of `field` across survivors is `<= max`.
185    NullRate {
186        field: String,
187        /// Maximum allowed null-or-missing proportion, in `[0.0, 1.0]`. Out-of-range values are rejected at compile time.
188        max: f64,
189        on_failure: OnFailure,
190    },
191    /// The composite `fields` tuple is unique across survivors.
192    Unique {
193        fields: Vec<String>,
194        on_failure: OnFailure,
195    },
196    /// Distinct values of `field` across survivors is within `[min, max]`.
197    DistinctCount {
198        field: String,
199        #[serde(default)]
200        min: Option<usize>,
201        #[serde(default)]
202        max: Option<usize>,
203        on_failure: OnFailure,
204    },
205}
206
207#[cfg(test)]
208mod tests {
209    use super::*;
210
211    #[test]
212    fn on_failure_serializes_snake_case() {
213        assert_eq!(
214            serde_json::to_string(&OnFailure::QuarantineBatch).unwrap(),
215            "\"quarantine_batch\""
216        );
217    }
218
219    #[test]
220    fn compare_op_round_trips() {
221        let op: CompareOp = serde_json::from_str("\"gte\"").unwrap();
222        assert_eq!(op, CompareOp::Gte);
223    }
224
225    #[test]
226    fn json_type_round_trips() {
227        let t: JsonType = serde_json::from_str("\"boolean\"").unwrap();
228        assert_eq!(t, JsonType::Boolean);
229    }
230
231    #[test]
232    fn parses_full_quality_block() {
233        let spec: QualitySpec = serde_json::from_value(serde_json::json!({
234            "record": [
235                { "type": "not_null", "field": "user_id", "on_failure": "quarantine" },
236                { "type": "compare", "field": "age", "op": "gte", "value": 0, "on_failure": "abort" },
237                { "type": "string_length", "field": "name", "min": 1, "max": 256, "on_failure": "quarantine" }
238            ],
239            "batch": [
240                { "type": "row_count", "min": 1, "max": 100000, "on_failure": "abort" },
241                { "type": "unique", "fields": ["id"], "on_failure": "quarantine" }
242            ]
243        }))
244        .unwrap();
245        assert_eq!(spec.record.len(), 3);
246        assert_eq!(spec.batch.len(), 2);
247        assert!(matches!(spec.record[0], RecordCheck::NotNull { .. }));
248        assert!(matches!(spec.batch[1], BatchCheck::Unique { .. }));
249        if let RecordCheck::NotNull {
250            treat_missing_as_null,
251            ..
252        } = &spec.record[0]
253        {
254            assert!(
255                *treat_missing_as_null,
256                "treat_missing_as_null defaults to true"
257            );
258        } else {
259            panic!("expected first record check to be NotNull");
260        }
261    }
262
263    #[test]
264    fn empty_quality_block_defaults_to_no_checks() {
265        let spec: QualitySpec = serde_json::from_str("{}").unwrap();
266        assert!(spec.record.is_empty());
267        assert!(spec.batch.is_empty());
268    }
269}
faucet_core/quality/config.rs

faucet_core/quality/
config.rs
