use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::fmt;
use dataprof_core::serde_helpers::rounded_2;
use dataprof_core::{ColumnProfile, DataType, ExecutionMetadata, QualityDimension};
use dataprof_metrics::{QualityAssessment, ScoreBounds, ScoreInterval};
use serde::Serialize;
use crate::profile_report::{
ColumnSelection, MetricSemantics, ProfileReport, QualityAnalysisStatus,
};
use crate::report_delta::{
ColumnChange, ColumnSide, ConfigurationComparison, RecordedHints, ReportDelta, ReportReference,
ReportSide, SchemaChange, Sides,
};
const PERCENTAGE_MAX: f64 = 100.0;
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub enum PolicyError {
#[non_exhaustive]
ThresholdOutOfRange {
code: CheckCode,
subject: Option<String>,
value: f64,
},
NoRequirements,
BaselineRequired,
}
impl fmt::Display for PolicyError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::ThresholdOutOfRange {
code,
subject,
value,
} => {
write!(f, "{code} threshold ")?;
if let Some(subject) = subject {
write!(f, "for `{subject}` ")?;
}
write!(
f,
"must be a percentage between 0 and {PERCENTAGE_MAX}, got {value}"
)
}
Self::NoRequirements => write!(
f,
"policy states no requirement; a gate must check something"
),
Self::BaselineRequired => write!(
f,
"policy states a rule about change, which needs a baseline report to compare \
against"
),
}
}
}
impl std::error::Error for PolicyError {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum PolicyScope {
#[default]
FullSource,
Observed,
}
impl fmt::Display for PolicyScope {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::FullSource => write!(f, "full_source"),
Self::Observed => write!(f, "observed"),
}
}
}
impl std::str::FromStr for PolicyScope {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"full_source" => Ok(Self::FullSource),
"observed" => Ok(Self::Observed),
_ => Err(format!(
"Unknown policy scope: {s}. Valid scopes: full_source, observed"
)),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum EvidenceGap {
Truncated,
Sampled,
RecordsSkipped,
QualitySampled,
CoverageUnrecorded,
}
impl fmt::Display for EvidenceGap {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Truncated => write!(f, "truncated"),
Self::Sampled => write!(f, "sampled"),
Self::RecordsSkipped => write!(f, "records_skipped"),
Self::QualitySampled => write!(f, "quality_sampled"),
Self::CoverageUnrecorded => write!(f, "coverage_unrecorded"),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(tag = "coverage", rename_all = "snake_case")]
#[non_exhaustive]
pub enum Evidence {
Complete,
#[non_exhaustive]
Incomplete {
reason: EvidenceGap,
},
}
impl Evidence {
pub fn is_complete(self) -> bool {
matches!(self, Self::Complete)
}
pub fn gap(self) -> Option<EvidenceGap> {
match self {
Self::Complete => None,
Self::Incomplete { reason } => Some(reason),
}
}
pub(crate) fn and(self, other: Self) -> Self {
match self {
Self::Complete => other,
incomplete => incomplete,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, serde::Serialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum CheckCode {
MinRows,
MinQualityScore,
MinDimensionScore,
MaxNullPercentage,
MaxNullCount,
MaxDuplicateRows,
RequireMetric,
MaxQualityScoreDrop,
MaxDimensionScoreDrop,
MaxNullPercentageIncrease,
RequireColumn,
ForbidRemovedColumn,
ForbidAddedColumn,
}
impl fmt::Display for CheckCode {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::MinRows => write!(f, "min_rows"),
Self::MinQualityScore => write!(f, "min_quality_score"),
Self::MinDimensionScore => write!(f, "min_dimension_score"),
Self::MaxNullPercentage => write!(f, "max_null_percentage"),
Self::MaxNullCount => write!(f, "max_null_count"),
Self::MaxDuplicateRows => write!(f, "max_duplicate_rows"),
Self::RequireMetric => write!(f, "require_metric"),
Self::MaxQualityScoreDrop => write!(f, "max_quality_score_drop"),
Self::MaxDimensionScoreDrop => write!(f, "max_dimension_score_drop"),
Self::MaxNullPercentageIncrease => write!(f, "max_null_percentage_increase"),
Self::RequireColumn => write!(f, "require_column"),
Self::ForbidRemovedColumn => write!(f, "forbid_removed_column"),
Self::ForbidAddedColumn => write!(f, "forbid_added_column"),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize)]
#[serde(untagged)]
#[non_exhaustive]
pub enum MetricValue {
Count(usize),
Percentage(#[serde(serialize_with = "dataprof_core::serde_helpers::round_2")] f64),
}
impl fmt::Display for MetricValue {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Count(count) => write!(f, "{count}"),
Self::Percentage(value) => write!(f, "{value:.2}"),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize)]
#[serde(tag = "comparison", rename_all = "snake_case")]
#[non_exhaustive]
pub enum Expectation {
#[non_exhaustive]
AtLeast {
value: MetricValue,
},
#[non_exhaustive]
AtMost {
value: MetricValue,
},
Analyzed,
#[non_exhaustive]
ChangeAtLeast {
value: MetricValue,
},
#[non_exhaustive]
ChangeAtMost {
value: MetricValue,
},
Present,
InBaseline,
}
impl fmt::Display for Expectation {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::AtLeast { value } => write!(f, "at least {value}"),
Self::AtMost { value } => write!(f, "at most {value}"),
Self::Analyzed => write!(f, "analyzed"),
Self::ChangeAtLeast { value } => write!(f, "a change of at least {value}"),
Self::ChangeAtMost { value } => write!(f, "a change of at most {value}"),
Self::Present => write!(f, "present"),
Self::InBaseline => write!(f, "in the baseline"),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(tag = "reason", rename_all = "snake_case")]
#[non_exhaustive]
pub enum NotEvaluated {
#[non_exhaustive]
QualityUnavailable {
quality_status: String,
},
NotAssessed,
ColumnNotProfiled,
#[non_exhaustive]
EvidenceIncomplete {
gap: EvidenceGap,
},
NotInBaseline,
SchemaChanged,
DimensionsChanged,
#[non_exhaustive]
DefinitionNotComparable {
definition: String,
},
#[non_exhaustive]
SettingChanged {
setting: String,
},
#[non_exhaustive]
SettingUnrecorded {
setting: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(tag = "status", rename_all = "snake_case")]
#[non_exhaustive]
pub enum CheckStatus {
Passed,
Failed,
NotEvaluated(NotEvaluated),
}
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
#[non_exhaustive]
pub struct Check {
pub code: CheckCode,
#[serde(skip_serializing_if = "Option::is_none")]
pub column: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub dimension: Option<QualityDimension>,
pub expected: Expectation,
#[serde(skip_serializing_if = "Option::is_none")]
pub observed: Option<MetricValue>,
pub scope: PolicyScope,
pub evidence: Evidence,
#[serde(skip_serializing_if = "Option::is_none")]
pub bounds: Option<CheckBounds>,
#[serde(skip_serializing_if = "Option::is_none")]
pub baseline: Option<BaselineReading>,
#[serde(flatten)]
pub status: CheckStatus,
pub message: String,
}
impl Check {
pub fn is_violation(&self) -> bool {
matches!(self.status, CheckStatus::Failed)
}
pub fn not_evaluated(&self) -> Option<&NotEvaluated> {
match &self.status {
CheckStatus::NotEvaluated(reason) => Some(reason),
_ => None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize)]
#[non_exhaustive]
pub struct BaselineReading {
pub observed: Option<MetricValue>,
pub evidence: Evidence,
pub change: Option<MetricValue>,
}
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
#[non_exhaustive]
pub struct BaselineComparison {
pub reports: Sides<ReportReference>,
pub schema: SchemaChange,
pub diagnostics: Vec<Diagnostic>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, serde::Serialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum DiagnosticCode {
ColumnAdded,
ColumnRemoved,
ColumnTypeChanged,
ColumnAllNull,
}
impl fmt::Display for DiagnosticCode {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::ColumnAdded => write!(f, "column_added"),
Self::ColumnRemoved => write!(f, "column_removed"),
Self::ColumnTypeChanged => write!(f, "column_type_changed"),
Self::ColumnAllNull => write!(f, "column_all_null"),
}
}
}
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
#[non_exhaustive]
pub struct Diagnostic {
pub code: DiagnosticCode,
pub column: String,
#[serde(
skip_serializing_if = "Option::is_none",
serialize_with = "serialize_data_types"
)]
pub data_type: Option<Sides<DataType>>,
#[serde(flatten)]
pub setting: Option<SettingDifference>,
pub message: String,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(tag = "reason", rename_all = "snake_case")]
#[non_exhaustive]
pub enum SettingDifference {
#[non_exhaustive]
SettingChanged {
setting: String,
},
#[non_exhaustive]
SettingUnrecorded {
setting: String,
},
}
fn serialize_data_types<S: serde::Serializer>(
types: &Option<Sides<DataType>>,
serializer: S,
) -> Result<S::Ok, S::Error> {
types
.as_ref()
.map(|types| Sides {
a: types.a.name(),
b: types.b.name(),
})
.serialize(serializer)
}
#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize)]
#[non_exhaustive]
pub struct CheckBounds {
#[serde(serialize_with = "dataprof_core::serde_helpers::round_2")]
pub lower: f64,
#[serde(serialize_with = "dataprof_core::serde_helpers::round_2")]
pub upper: f64,
pub confidence_level: f64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum Verdict {
Pass,
Fail,
Inconclusive,
}
impl fmt::Display for Verdict {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Pass => write!(f, "pass"),
Self::Fail => write!(f, "fail"),
Self::Inconclusive => write!(f, "inconclusive"),
}
}
}
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
#[non_exhaustive]
pub struct GateResult {
pub verdict: Verdict,
pub scope: PolicyScope,
pub evidence: Evidence,
pub checks: Vec<Check>,
#[serde(skip_serializing_if = "Option::is_none")]
pub baseline: Option<BaselineComparison>,
}
impl GateResult {
pub fn passed(&self) -> bool {
self.verdict == Verdict::Pass
}
pub fn violations(&self) -> impl Iterator<Item = &Check> {
self.checks.iter().filter(|check| check.is_violation())
}
pub fn unevaluated(&self) -> impl Iterator<Item = &Check> {
self.checks
.iter()
.filter(|check| check.not_evaluated().is_some())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum RequiredMetric {
Quality,
Dimension(QualityDimension),
}
#[derive(Debug, Clone, Default)]
#[non_exhaustive]
pub struct QualityPolicy {
min_rows: Option<usize>,
min_quality_score: Option<f64>,
min_dimension_scores: BTreeMap<String, (QualityDimension, f64)>,
max_null_percentage: BTreeMap<String, f64>,
max_null_percentage_any: Option<f64>,
max_null_count: BTreeMap<String, usize>,
max_null_count_any: Option<usize>,
max_duplicate_rows: Option<usize>,
require_metrics: Vec<RequiredMetric>,
max_quality_score_drop: Option<f64>,
max_dimension_score_drops: BTreeMap<String, (QualityDimension, f64)>,
max_null_percentage_increase: BTreeMap<String, f64>,
max_null_percentage_increase_any: Option<f64>,
require_columns: BTreeSet<String>,
forbid_removed_columns: bool,
forbid_added_columns: bool,
scope: PolicyScope,
}
impl QualityPolicy {
pub fn new() -> Self {
Self::default()
}
pub fn min_rows(mut self, min: usize) -> Self {
self.min_rows = Some(min);
self
}
pub fn min_quality_score(mut self, min: f64) -> Self {
self.min_quality_score = Some(min);
self
}
pub fn min_dimension_score(mut self, dimension: QualityDimension, min: f64) -> Self {
self.min_dimension_scores
.insert(dimension.to_string(), (dimension, min));
self
}
pub fn max_null_percentage(mut self, column: impl Into<String>, max: f64) -> Self {
self.max_null_percentage.insert(column.into(), max);
self
}
pub fn max_null_percentage_any(mut self, max: f64) -> Self {
self.max_null_percentage_any = Some(max);
self
}
pub fn max_null_count(mut self, column: impl Into<String>, max: usize) -> Self {
self.max_null_count.insert(column.into(), max);
self
}
pub fn max_null_count_any(mut self, max: usize) -> Self {
self.max_null_count_any = Some(max);
self
}
pub fn max_duplicate_rows(mut self, max: usize) -> Self {
self.max_duplicate_rows = Some(max);
self
}
pub fn require_quality(mut self) -> Self {
self.require_metrics.push(RequiredMetric::Quality);
self
}
pub fn require_dimension(mut self, dimension: QualityDimension) -> Self {
self.require_metrics
.push(RequiredMetric::Dimension(dimension));
self
}
pub fn max_quality_score_drop(mut self, points: f64) -> Self {
self.max_quality_score_drop = Some(points);
self
}
pub fn max_dimension_score_drop(mut self, dimension: QualityDimension, points: f64) -> Self {
self.max_dimension_score_drops
.insert(dimension.to_string(), (dimension, points));
self
}
pub fn max_null_percentage_increase(mut self, column: impl Into<String>, points: f64) -> Self {
self.max_null_percentage_increase
.insert(column.into(), points);
self
}
pub fn max_null_percentage_increase_any(mut self, points: f64) -> Self {
self.max_null_percentage_increase_any = Some(points);
self
}
pub fn require_columns<I, S>(mut self, columns: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
self.require_columns
.extend(columns.into_iter().map(Into::into));
self
}
pub fn forbid_removed_columns(mut self) -> Self {
self.forbid_removed_columns = true;
self
}
pub fn forbid_added_columns(mut self) -> Self {
self.forbid_added_columns = true;
self
}
pub fn scope(mut self, scope: PolicyScope) -> Self {
self.scope = scope;
self
}
pub fn validate(&self) -> Result<(), PolicyError> {
if let Some(min) = self.min_quality_score {
check_percentage(CheckCode::MinQualityScore, None, min)?;
}
for (dimension, min) in self.min_dimension_scores.values() {
check_percentage(
CheckCode::MinDimensionScore,
Some(dimension.to_string()),
*min,
)?;
}
for (column, max) in &self.max_null_percentage {
check_percentage(CheckCode::MaxNullPercentage, Some(column.clone()), *max)?;
}
if let Some(max) = self.max_null_percentage_any {
check_percentage(CheckCode::MaxNullPercentage, None, max)?;
}
if let Some(points) = self.max_quality_score_drop {
check_percentage(CheckCode::MaxQualityScoreDrop, None, points)?;
}
for (dimension, points) in self.max_dimension_score_drops.values() {
check_percentage(
CheckCode::MaxDimensionScoreDrop,
Some(dimension.to_string()),
*points,
)?;
}
for (column, points) in &self.max_null_percentage_increase {
check_percentage(
CheckCode::MaxNullPercentageIncrease,
Some(column.clone()),
*points,
)?;
}
if let Some(points) = self.max_null_percentage_increase_any {
check_percentage(CheckCode::MaxNullPercentageIncrease, None, points)?;
}
if self.is_empty() {
return Err(PolicyError::NoRequirements);
}
Ok(())
}
fn is_empty(&self) -> bool {
self.min_rows.is_none()
&& self.min_quality_score.is_none()
&& self.min_dimension_scores.is_empty()
&& self.max_null_percentage.is_empty()
&& self.max_null_percentage_any.is_none()
&& self.max_null_count.is_empty()
&& self.max_null_count_any.is_none()
&& self.max_duplicate_rows.is_none()
&& self.require_metrics.is_empty()
&& self.require_columns.is_empty()
&& !self.has_change_rules()
}
fn has_change_rules(&self) -> bool {
self.max_quality_score_drop.is_some()
|| !self.max_dimension_score_drops.is_empty()
|| !self.max_null_percentage_increase.is_empty()
|| self.max_null_percentage_increase_any.is_some()
|| self.forbid_removed_columns
|| self.forbid_added_columns
}
pub fn evaluate(&self, report: &ProfileReport) -> Result<GateResult, PolicyError> {
self.validate()?;
if self.has_change_rules() {
return Err(PolicyError::BaselineRequired);
}
let checks = self.absolute_checks(report);
Ok(GateResult {
verdict: verdict_of(&checks),
scope: self.scope,
evidence: scan_evidence(&report.execution),
checks,
baseline: None,
})
}
pub fn evaluate_against(
&self,
report: &ProfileReport,
baseline: &ReportSide,
) -> Result<GateResult, PolicyError> {
self.validate()?;
let current = ReportSide::from_report(report);
let delta = ReportDelta::between_sides(baseline, ¤t);
let identifiers = Sides {
a: baseline.identifier_order(),
b: current.identifier_order(),
};
let mut checks = self.absolute_checks(report);
if let Some(points) = self.max_quality_score_drop {
checks.push(self.score_drop_check(report, &delta, &identifiers, None, points));
}
for dimension in QualityDimension::all() {
if let Some((_, points)) = self.max_dimension_score_drops.get(&dimension.to_string()) {
checks.push(self.score_drop_check(
report,
&delta,
&identifiers,
Some(dimension),
*points,
));
}
}
checks.extend(self.null_increase_checks(&delta));
let (baseline_names, current_names) = (column_names(baseline), column_names(¤t));
if self.forbid_removed_columns {
for column in baseline.columns() {
checks.push(self.forbid_removed_check(
baseline,
¤t,
¤t_names,
&column.name,
));
}
}
if self.forbid_added_columns {
for column in current.columns() {
checks.push(self.forbid_added_check(
baseline,
¤t,
&baseline_names,
&column.name,
));
}
if self.scope == PolicyScope::FullSource
&& let Some(reason) = unenumerated_columns(¤t)
{
checks.push(self.unprofiled_columns_check(baseline, ¤t, reason));
}
}
let diagnostics = schema_diagnostics(baseline, ¤t, &delta);
Ok(GateResult {
verdict: verdict_of(&checks),
scope: self.scope,
evidence: scan_evidence(&report.execution),
checks,
baseline: Some(BaselineComparison {
reports: delta.reports,
schema: delta.schema,
diagnostics,
}),
})
}
fn absolute_checks(&self, report: &ProfileReport) -> Vec<Check> {
let scan = scan_evidence(&report.execution);
let mut checks = Vec::new();
for metric in required_metrics_in_order(&self.require_metrics) {
checks.push(self.require_metric_check(report, metric));
}
if !self.require_columns.is_empty() {
let side = ReportSide::from_report(report);
let names = column_names(&side);
for column in &self.require_columns {
checks.push(self.require_column_check(&side, &names, column));
}
}
if let Some(min) = self.min_rows {
checks.push(self.row_count_check(report, scan, min));
}
if let Some(min) = self.min_quality_score {
checks.push(self.quality_score_check(report, scan, min));
}
for dimension in QualityDimension::all() {
if let Some((_, min)) = self.min_dimension_scores.get(&dimension.to_string()) {
checks.push(self.dimension_score_check(report, scan, dimension, *min));
}
}
checks.extend(self.null_percentage_checks(report, scan));
checks.extend(self.null_count_checks(report, scan));
if let Some(max) = self.max_duplicate_rows {
checks.push(self.duplicate_rows_check(report, scan, max));
}
checks
}
fn decide_aggregate(&self, evidence: Evidence, satisfied: bool) -> CheckStatus {
match (self.scope, evidence.gap()) {
(PolicyScope::FullSource, Some(gap)) => {
CheckStatus::NotEvaluated(NotEvaluated::EvidenceIncomplete { gap })
}
_ if satisfied => CheckStatus::Passed,
_ => CheckStatus::Failed,
}
}
fn require_metric_check(&self, report: &ProfileReport, metric: RequiredMetric) -> Check {
let dimension = match metric {
RequiredMetric::Quality => None,
RequiredMetric::Dimension(dimension) => Some(dimension),
};
let analyzed = match (report.quality.as_ref(), dimension) {
(None, _) => false,
(Some(_), None) => true,
(Some(quality), Some(dimension)) => {
quality.metrics.dimension_score(dimension).is_some()
}
};
let message = match (analyzed, dimension) {
(true, None) => "quality was analyzed",
(false, None) => "the report carries no quality assessment",
(true, Some(_)) => "the required dimension was assessed",
(false, Some(_)) => "the required dimension was not assessed",
};
Check {
code: CheckCode::RequireMetric,
column: None,
dimension,
expected: Expectation::Analyzed,
observed: None,
scope: self.scope,
evidence: Evidence::Complete,
bounds: None,
baseline: None,
status: if analyzed {
CheckStatus::Passed
} else {
CheckStatus::Failed
},
message: message.to_string(),
}
}
fn row_count_check(&self, report: &ProfileReport, scan: Evidence, min: usize) -> Check {
let observed = report.execution.rows_processed;
let status = match (self.scope, scan.gap()) {
_ if observed >= min => CheckStatus::Passed,
(PolicyScope::FullSource, Some(gap)) => {
CheckStatus::NotEvaluated(NotEvaluated::EvidenceIncomplete { gap })
}
_ => CheckStatus::Failed,
};
let message = match status {
CheckStatus::Passed => "the scan read at least the required number of rows",
CheckStatus::Failed => "fewer rows were read than the required minimum",
CheckStatus::NotEvaluated(_) => {
"fewer rows were read than the required minimum, and the rows that were not \
read may make up the difference"
}
};
Check {
code: CheckCode::MinRows,
column: None,
dimension: None,
expected: Expectation::AtLeast {
value: MetricValue::Count(min),
},
observed: Some(MetricValue::Count(observed)),
scope: self.scope,
evidence: scan,
bounds: None,
baseline: None,
status,
message: message.to_string(),
}
}
fn quality_score_check(&self, report: &ProfileReport, scan: Evidence, min: f64) -> Check {
let mut check = Check {
code: CheckCode::MinQualityScore,
column: None,
dimension: None,
expected: Expectation::AtLeast {
value: MetricValue::Percentage(min),
},
observed: None,
scope: self.scope,
evidence: scan.and(quality_evidence(
report.quality.as_ref(),
Provenance::Overall,
)),
bounds: None,
baseline: None,
status: CheckStatus::Passed,
message: String::new(),
};
let Some(quality) = report.quality.as_ref() else {
return unavailable_quality(check, report);
};
let Some(score) = quality.score() else {
check.status = CheckStatus::NotEvaluated(NotEvaluated::NotAssessed);
check.message =
"no quality dimension had anything to assess, so there is no overall score"
.to_string();
return check;
};
check.observed = Some(MetricValue::Percentage(score));
let bounds = quality
.score_bounds()
.and_then(|bounds| check_bounds(bounds, bounds.overall_score));
self.decide_score(&mut check, score, min, bounds, "the overall quality score");
check
}
fn dimension_score_check(
&self,
report: &ProfileReport,
scan: Evidence,
dimension: QualityDimension,
min: f64,
) -> Check {
let mut check = Check {
code: CheckCode::MinDimensionScore,
column: None,
dimension: Some(dimension),
expected: Expectation::AtLeast {
value: MetricValue::Percentage(min),
},
observed: None,
scope: self.scope,
evidence: scan.and(quality_evidence(
report.quality.as_ref(),
Provenance::Dimension(dimension),
)),
bounds: None,
baseline: None,
status: CheckStatus::Passed,
message: String::new(),
};
let Some(quality) = report.quality.as_ref() else {
return unavailable_quality(check, report);
};
let Some(score) = quality
.scores()
.dimension_scores
.get(&dimension.to_string())
.copied()
.flatten()
else {
check.status = CheckStatus::NotEvaluated(NotEvaluated::NotAssessed);
check.message = "this dimension had nothing to assess in this run".to_string();
return check;
};
check.observed = Some(MetricValue::Percentage(score));
let bounds = quality.score_bounds().and_then(|bounds| {
check_bounds(
bounds,
bounds
.dimension_scores
.get(&dimension.to_string())
.copied()
.flatten(),
)
});
self.decide_score(&mut check, score, min, bounds, "this dimension's score");
check
}
fn decide_score(
&self,
check: &mut Check,
score: f64,
min: f64,
bounds: Option<CheckBounds>,
subject: &str,
) {
let sampled = check.evidence.gap() == Some(EvidenceGap::QualitySampled);
let min = rounded_2(min);
match bounds {
Some(bounds) if self.scope == PolicyScope::FullSource && sampled => {
check.bounds = Some(bounds);
check.status = if rounded_2(bounds.lower) >= min {
CheckStatus::Passed
} else if rounded_2(bounds.upper) < min {
CheckStatus::Failed
} else {
CheckStatus::NotEvaluated(NotEvaluated::EvidenceIncomplete {
gap: EvidenceGap::QualitySampled,
})
};
check.message = bounded_message(&check.status, subject);
}
_ => {
check.status = self.decide_aggregate(check.evidence, rounded_2(score) >= min);
check.message = aggregate_message(&check.status, subject);
}
}
}
fn null_percentage_checks(&self, report: &ProfileReport, scan: Evidence) -> Vec<Check> {
let mut checks = Vec::new();
for (column, max) in &self.max_null_percentage {
let profile = report
.column_profiles
.iter()
.find(|candidate| &candidate.name == column);
checks.push(self.null_percentage_check(column, profile, scan, *max));
}
if let Some(max) = self.max_null_percentage_any {
for profile in &report.column_profiles {
if !self.max_null_percentage.contains_key(&profile.name) {
checks.push(self.null_percentage_check(
&profile.name,
Some(profile),
scan,
max,
));
}
}
}
checks
}
fn null_percentage_check(
&self,
column: &str,
profile: Option<&ColumnProfile>,
scan: Evidence,
max: f64,
) -> Check {
let mut check = Check {
code: CheckCode::MaxNullPercentage,
column: Some(column.to_string()),
dimension: None,
expected: Expectation::AtMost {
value: MetricValue::Percentage(max),
},
observed: None,
scope: self.scope,
evidence: scan,
bounds: None,
baseline: None,
status: CheckStatus::Passed,
message: String::new(),
};
let Some(profile) = profile else {
check.status = CheckStatus::NotEvaluated(NotEvaluated::ColumnNotProfiled);
check.message = "this column has no profile in the report".to_string();
return check;
};
let Some(percentage) = null_percentage(profile) else {
check.status = CheckStatus::NotEvaluated(NotEvaluated::NotAssessed);
check.message = "no values were read for this column".to_string();
return check;
};
check.observed = Some(MetricValue::Percentage(percentage));
check.status =
self.decide_aggregate(check.evidence, rounded_2(percentage) <= rounded_2(max));
check.message = match check.status {
CheckStatus::Passed => "this column's null percentage is within the allowance",
CheckStatus::Failed => "this column's null percentage is above the allowance",
CheckStatus::NotEvaluated(_) => {
"the scan does not cover the whole source, and a null percentage over \
part of it bounds nothing about the rest"
}
}
.to_string();
check
}
fn null_count_checks(&self, report: &ProfileReport, scan: Evidence) -> Vec<Check> {
let mut checks = Vec::new();
for (column, max) in &self.max_null_count {
let profile = report
.column_profiles
.iter()
.find(|candidate| &candidate.name == column);
checks.push(self.null_count_check(column, profile, scan, *max));
}
if let Some(max) = self.max_null_count_any {
for profile in &report.column_profiles {
if !self.max_null_count.contains_key(&profile.name) {
checks.push(self.null_count_check(&profile.name, Some(profile), scan, max));
}
}
}
checks
}
fn null_count_check(
&self,
column: &str,
profile: Option<&ColumnProfile>,
scan: Evidence,
max: usize,
) -> Check {
let mut check = Check {
code: CheckCode::MaxNullCount,
column: Some(column.to_string()),
dimension: None,
expected: Expectation::AtMost {
value: MetricValue::Count(max),
},
observed: None,
scope: self.scope,
evidence: scan,
bounds: None,
baseline: None,
status: CheckStatus::Passed,
message: String::new(),
};
let Some(profile) = profile else {
check.status = CheckStatus::NotEvaluated(NotEvaluated::ColumnNotProfiled);
check.message = "this column has no profile in the report".to_string();
return check;
};
let exceeded = profile.null_count > max;
check.observed = Some(MetricValue::Count(profile.null_count));
check.status = match (self.scope, scan.gap()) {
(PolicyScope::FullSource, Some(_)) if exceeded => CheckStatus::Failed,
(PolicyScope::FullSource, Some(gap)) => {
CheckStatus::NotEvaluated(NotEvaluated::EvidenceIncomplete { gap })
}
_ if exceeded => CheckStatus::Failed,
_ => CheckStatus::Passed,
};
check.message = match check.status {
CheckStatus::Passed => "this column's null count is within the allowance",
CheckStatus::Failed => "nulls were observed in this column above the allowance",
CheckStatus::NotEvaluated(_) => {
"no nulls above the allowance were observed in this column, and the rows \
that were not read may hold more"
}
}
.to_string();
check
}
fn duplicate_rows_check(&self, report: &ProfileReport, scan: Evidence, max: usize) -> Check {
let mut check = Check {
code: CheckCode::MaxDuplicateRows,
column: None,
dimension: Some(QualityDimension::Uniqueness),
expected: Expectation::AtMost {
value: MetricValue::Count(max),
},
observed: None,
scope: self.scope,
evidence: scan.and(quality_evidence(
report.quality.as_ref(),
Provenance::Component("duplicate_rows"),
)),
bounds: None,
baseline: None,
status: CheckStatus::Passed,
message: String::new(),
};
let Some(quality) = report.quality.as_ref() else {
return unavailable_quality(check, report);
};
let Some(uniqueness) = quality
.metrics
.uniqueness
.as_ref()
.filter(|metrics| metrics.rows_checked > 0)
else {
check.status = CheckStatus::NotEvaluated(NotEvaluated::NotAssessed);
check.message = "no rows were scanned for duplicates in this run".to_string();
return check;
};
let observed = uniqueness.duplicate_rows;
let exceeded = observed > max;
let estimated = uniqueness.duplicate_rows_approximate;
check.observed = Some(MetricValue::Count(observed));
check.status = match (self.scope, check.evidence.gap()) {
(PolicyScope::FullSource, Some(_)) if exceeded && !estimated => CheckStatus::Failed,
(PolicyScope::FullSource, Some(gap)) => {
CheckStatus::NotEvaluated(NotEvaluated::EvidenceIncomplete { gap })
}
_ if exceeded => CheckStatus::Failed,
_ => CheckStatus::Passed,
};
check.message = match check.status {
CheckStatus::Passed => "the duplicate-row count is within the allowance",
CheckStatus::Failed if estimated => {
"the estimated duplicate-row count is above the allowance"
}
CheckStatus::Failed => "duplicate rows were observed above the allowance",
CheckStatus::NotEvaluated(_) if estimated => {
"the duplicate-row count is an estimate, which can neither witness \
duplicates above the allowance nor rule them out"
}
CheckStatus::NotEvaluated(_) => {
"no duplicate above the allowance was observed, and the rows that were not \
read may hold more"
}
}
.to_string();
check
}
fn score_drop_check(
&self,
report: &ProfileReport,
delta: &ReportDelta,
identifiers: &Sides<Option<&[String]>>,
dimension: Option<QualityDimension>,
points: f64,
) -> Check {
let (code, change, subject) = match dimension {
None => (
CheckCode::MaxQualityScoreDrop,
&delta.quality_score,
"the overall quality score",
),
Some(dimension) => (
CheckCode::MaxDimensionScoreDrop,
&delta
.dimensions
.iter()
.find(|change| change.dimension == dimension)
.expect("a delta lists every dimension this build knows")
.score,
"this dimension's score",
),
};
let mut check = Check {
code,
column: None,
dimension,
expected: Expectation::ChangeAtLeast {
value: MetricValue::Percentage(0.0 - points),
},
observed: change.b.map(MetricValue::Percentage),
scope: self.scope,
evidence: change.evidence.b,
bounds: None,
baseline: Some(baseline_reading(change.a, change.evidence.a, change.abs)),
status: CheckStatus::Passed,
message: String::new(),
};
if report.quality.is_none() {
return unavailable_quality(check, report);
}
let schema_changed = !delta.schema.added.is_empty() || !delta.schema.removed.is_empty();
let dimensions_changed = dimension.is_none()
&& delta
.dimensions
.iter()
.any(|change| change.score.a.is_some() != change.score.b.is_some());
let reason = if change.b.is_none() {
Some(NotEvaluated::NotAssessed)
} else if change.a.is_none() {
Some(NotEvaluated::NotInBaseline)
} else if schema_changed {
Some(NotEvaluated::SchemaChanged)
} else if dimensions_changed {
Some(NotEvaluated::DimensionsChanged)
} else {
definition_gap(
delta.metric_semantics.a.as_ref(),
delta.metric_semantics.b.as_ref(),
score_definitions(dimension),
)
.or_else(|| setting_gap(&delta.configuration, identifiers, dimension))
};
if let Some(reason) = reason {
return not_evaluated_change(check, reason, subject);
}
let change_points = change
.abs
.expect("a delta has a change whenever both sides have a score");
check.status =
self.decide_change(&change.evidence, change_points >= 0.0 - rounded_2(points));
check.message = match check.status {
CheckStatus::Passed => {
format!("{subject} dropped no more than allowed from the baseline")
}
CheckStatus::Failed => format!("{subject} dropped more than allowed from the baseline"),
CheckStatus::NotEvaluated(_) => evidence_message(subject),
};
check
}
fn null_increase_checks(&self, delta: &ReportDelta) -> Vec<Check> {
let index = ColumnIndex::new(delta);
let mut checks = Vec::new();
for (column, points) in &self.max_null_percentage_increase {
checks.push(self.null_increase_check(delta, &index, column, *points));
}
if let Some(points) = self.max_null_percentage_increase_any {
for column in &delta.schema.common {
if !self.max_null_percentage_increase.contains_key(column) {
checks.push(self.null_increase_check(delta, &index, column, points));
}
}
}
checks
}
fn null_increase_check(
&self,
delta: &ReportDelta,
index: &ColumnIndex<'_>,
column: &str,
points: f64,
) -> Check {
let change = index.changes.get(column).copied();
let (null_pct_a, null_pct_b, null_pct_delta) = change.map_or(
(None, None, None),
|ColumnChange {
null_pct_a,
null_pct_b,
null_pct_delta,
..
}| (*null_pct_a, *null_pct_b, *null_pct_delta),
);
let in_report = index.in_report.contains(column);
let in_baseline = index.in_baseline.contains(column);
let evidence = Sides {
a: delta.reports.a.evidence,
b: delta.reports.b.evidence,
};
let mut check = Check {
code: CheckCode::MaxNullPercentageIncrease,
column: Some(column.to_string()),
dimension: None,
expected: Expectation::ChangeAtMost {
value: MetricValue::Percentage(points),
},
observed: null_pct_b.map(MetricValue::Percentage),
scope: self.scope,
evidence: evidence.b,
bounds: None,
baseline: Some(baseline_reading(null_pct_a, evidence.a, null_pct_delta)),
status: CheckStatus::Passed,
message: String::new(),
};
let reason = if !in_report {
Some(NotEvaluated::ColumnNotProfiled)
} else if !in_baseline {
Some(NotEvaluated::NotInBaseline)
} else if null_pct_b.is_none() {
Some(NotEvaluated::NotAssessed)
} else if null_pct_a.is_none() {
Some(NotEvaluated::NotInBaseline)
} else {
definition_gap(
delta.metric_semantics.a.as_ref(),
delta.metric_semantics.b.as_ref(),
&[Definition::NullTokens],
)
};
let subject = "this column's null percentage";
if let Some(reason) = reason {
return not_evaluated_change(check, reason, subject);
}
let change_points =
null_pct_delta.expect("a delta has a change whenever both sides have a value");
check.status = self.decide_change(&evidence, change_points <= rounded_2(points));
check.message = match check.status {
CheckStatus::Passed => {
format!("{subject} rose no more than allowed above the baseline")
}
CheckStatus::Failed => format!("{subject} rose more than allowed above the baseline"),
CheckStatus::NotEvaluated(_) => evidence_message(subject),
};
check
}
fn absence_status(
&self,
lacking: &ReportSide,
column: &str,
unselected: NotEvaluated,
) -> CheckStatus {
match lacking.column_selection() {
None => CheckStatus::NotEvaluated(NotEvaluated::SettingUnrecorded {
setting: "columns".to_string(),
}),
Some(selection) if !selection.includes(column) => CheckStatus::NotEvaluated(unselected),
Some(_) => match (self.scope, lacking.schema_evidence().gap()) {
(PolicyScope::FullSource, Some(gap)) => {
CheckStatus::NotEvaluated(NotEvaluated::EvidenceIncomplete { gap })
}
_ => CheckStatus::Failed,
},
}
}
fn require_column_check(
&self,
current: &ReportSide,
current_names: &HashSet<&str>,
column: &str,
) -> Check {
let status = if current_names.contains(column) {
CheckStatus::Passed
} else {
self.absence_status(current, column, NotEvaluated::ColumnNotProfiled)
};
let message = match &status {
CheckStatus::Passed => "the required column was profiled",
CheckStatus::Failed => "the required column is not in the source the report read",
CheckStatus::NotEvaluated(NotEvaluated::ColumnNotProfiled) => {
"the run's column selection left the required column out"
}
CheckStatus::NotEvaluated(NotEvaluated::SettingUnrecorded { .. }) => {
"the report does not record which columns it selected, so a missing column \
cannot be shown to be absent from the source"
}
CheckStatus::NotEvaluated(_) => {
"the required column is not in the records read, and the records that were not \
read may hold it"
}
};
self.column_check(
CheckCode::RequireColumn,
column,
Expectation::Present,
current,
None,
status,
message,
)
}
fn forbid_removed_check(
&self,
baseline: &ReportSide,
current: &ReportSide,
current_names: &HashSet<&str>,
column: &str,
) -> Check {
let status = if current_names.contains(column) {
CheckStatus::Passed
} else {
self.absence_status(
current,
column,
NotEvaluated::SettingChanged {
setting: "columns".to_string(),
},
)
};
let message = match &status {
CheckStatus::Passed => "the column the baseline profiled is still profiled",
CheckStatus::Failed => {
"a column the baseline profiled is not in the source the report read"
}
CheckStatus::NotEvaluated(NotEvaluated::SettingChanged { .. }) => {
"the report's column selection left out a column the baseline profiled"
}
CheckStatus::NotEvaluated(NotEvaluated::SettingUnrecorded { .. }) => {
"the report does not record which columns it selected, so a column the baseline \
profiled cannot be shown to be absent from its source"
}
CheckStatus::NotEvaluated(_) => {
"a column the baseline profiled is not in the records the report read, and the \
records that were not read may hold it"
}
};
self.column_check(
CheckCode::ForbidRemovedColumn,
column,
Expectation::Present,
current,
Some(baseline),
status,
message,
)
}
fn forbid_added_check(
&self,
baseline: &ReportSide,
current: &ReportSide,
baseline_names: &HashSet<&str>,
column: &str,
) -> Check {
let status = if baseline_names.contains(column) {
CheckStatus::Passed
} else {
self.absence_status(
baseline,
column,
NotEvaluated::SettingChanged {
setting: "columns".to_string(),
},
)
};
let message = match &status {
CheckStatus::Passed => "the column is one the baseline profiled",
CheckStatus::Failed => {
"the report profiled a column the baseline's source did not have"
}
CheckStatus::NotEvaluated(NotEvaluated::SettingChanged { .. }) => {
"the baseline's column selection left this column out"
}
CheckStatus::NotEvaluated(NotEvaluated::SettingUnrecorded { .. }) => {
"the baseline does not record which columns it selected, so this column cannot be \
shown to be new"
}
CheckStatus::NotEvaluated(_) => {
"this column is not in the records the baseline read, and the records it did not \
read may hold it"
}
};
self.column_check(
CheckCode::ForbidAddedColumn,
column,
Expectation::InBaseline,
current,
Some(baseline),
status,
message,
)
}
fn unprofiled_columns_check(
&self,
baseline: &ReportSide,
current: &ReportSide,
reason: NotEvaluated,
) -> Check {
let message = match &reason {
NotEvaluated::ColumnNotProfiled => {
"the run profiled a column selection, which may have left out a source column \
the baseline lacked"
}
NotEvaluated::SettingUnrecorded { .. } => {
"the report does not record which columns it selected, so its columns may not \
be all of its source's"
}
_ => {
"the report read part of its source, and the records it did not read may hold \
a column the baseline lacked"
}
};
Check {
column: None,
..self.column_check(
CheckCode::ForbidAddedColumn,
"",
Expectation::InBaseline,
current,
Some(baseline),
CheckStatus::NotEvaluated(reason),
message,
)
}
}
#[allow(clippy::too_many_arguments)]
fn column_check(
&self,
code: CheckCode,
column: &str,
expected: Expectation,
current: &ReportSide,
baseline: Option<&ReportSide>,
status: CheckStatus,
message: &str,
) -> Check {
Check {
code,
column: Some(column.to_string()),
dimension: None,
expected,
observed: None,
scope: self.scope,
evidence: current.schema_evidence(),
bounds: None,
baseline: baseline.map(|baseline| BaselineReading {
observed: None,
evidence: baseline.schema_evidence(),
change: None,
}),
status,
message: message.to_string(),
}
}
fn decide_change(&self, evidence: &Sides<Evidence>, satisfied: bool) -> CheckStatus {
match (self.scope, evidence.b.gap().or(evidence.a.gap())) {
(PolicyScope::FullSource, Some(gap)) => {
CheckStatus::NotEvaluated(NotEvaluated::EvidenceIncomplete { gap })
}
_ if satisfied => CheckStatus::Passed,
_ => CheckStatus::Failed,
}
}
}
fn schema_diagnostics(
baseline: &ReportSide,
current: &ReportSide,
delta: &ReportDelta,
) -> Vec<Diagnostic> {
let diagnostic = |code, column: &str, data_type, message: &str| Diagnostic {
code,
column: column.to_string(),
data_type,
setting: None,
message: message.to_string(),
};
let identified = |side: &ReportSide, column: &str| {
side.identifier_order()
.map(|names| names.iter().any(|name| name == column))
};
fn names(names: &[String]) -> HashSet<&str> {
names.iter().map(String::as_str).collect()
}
fn by_name(side: &ReportSide) -> HashMap<&str, &ColumnSide> {
side.columns()
.iter()
.map(|column| (column.name.as_str(), column))
.collect()
}
let (added, removed) = (names(&delta.schema.added), names(&delta.schema.removed));
let (baseline_columns, current_columns) = (by_name(baseline), by_name(current));
delta
.columns
.iter()
.filter_map(|change| {
let column = change.name.as_str();
let unselected = |side: &ReportSide| match side.column_selection() {
Some(selection) if selection.includes(column) => None,
Some(_) => Some(SettingDifference::SettingChanged {
setting: "columns".to_string(),
}),
None => Some(SettingDifference::SettingUnrecorded {
setting: "columns".to_string(),
}),
};
if added.contains(column) {
let setting = unselected(baseline);
let message = match setting {
None => {
"the report profiled a column the baseline did not find in its source, \
so the baseline has nothing to measure its change from"
}
Some(_) => {
"the report profiled a column the baseline has no profile for, and the \
baseline's column selection is not shown to include it"
}
};
return Some(Diagnostic {
setting,
..diagnostic(DiagnosticCode::ColumnAdded, column, None, message)
});
}
if removed.contains(column) {
let setting = unselected(current);
let message = match setting {
None => "the baseline profiled a column the report did not find in its source",
Some(_) => {
"the baseline profiled a column the report has no profile for, and the \
report's column selection is not shown to include it"
}
};
return Some(Diagnostic {
setting,
..diagnostic(DiagnosticCode::ColumnRemoved, column, None, message)
});
}
let (a, b) = (
*baseline_columns.get(column)?,
*current_columns.get(column)?,
);
if a.populated()? && b.all_null()? {
return Some(diagnostic(
DiagnosticCode::ColumnAllNull,
column,
None,
"this column held values in the baseline, and every value read for it in \
the report is null",
));
}
match (&a.data_type, &b.data_type) {
(Some(type_a), Some(type_b))
if type_a != type_b && a.populated()? && b.populated()? =>
{
let mut retyped = diagnostic(
DiagnosticCode::ColumnTypeChanged,
column,
Some(Sides {
a: type_a.clone(),
b: type_b.clone(),
}),
"this column holds values on both sides, and its inferred type differs",
);
if *type_a == DataType::Identifier || *type_b == DataType::Identifier {
let setting = "identifier_columns".to_string();
retyped.setting =
match (identified(baseline, column), identified(current, column)) {
(Some(a), Some(b)) if a == b => None,
(Some(_), Some(_)) => {
Some(SettingDifference::SettingChanged { setting })
}
_ => Some(SettingDifference::SettingUnrecorded { setting }),
};
if retyped.setting.is_some() {
retyped.message = "this column holds values on both sides, and its \
inferred type differs; the identifier hint, \
which sets the type, is not shown to be the \
same on both sides"
.to_string();
}
}
Some(retyped)
}
_ => None,
}
})
.collect()
}
struct ColumnIndex<'a> {
changes: HashMap<&'a str, &'a ColumnChange>,
in_report: HashSet<&'a str>,
in_baseline: HashSet<&'a str>,
}
impl<'a> ColumnIndex<'a> {
fn new(delta: &'a ReportDelta) -> Self {
let schema = &delta.schema;
let names = |lists: [&'a [String]; 2]| {
lists
.into_iter()
.flatten()
.map(String::as_str)
.collect::<HashSet<_>>()
};
Self {
changes: delta
.columns
.iter()
.map(|change| (change.name.as_str(), change))
.collect(),
in_report: names([&schema.common, &schema.added]),
in_baseline: names([&schema.common, &schema.removed]),
}
}
}
fn unenumerated_columns(side: &ReportSide) -> Option<NotEvaluated> {
match side.column_selection() {
None => Some(NotEvaluated::SettingUnrecorded {
setting: "columns".to_string(),
}),
Some(ColumnSelection::All) => side
.schema_evidence()
.gap()
.map(|gap| NotEvaluated::EvidenceIncomplete { gap }),
Some(_) => Some(NotEvaluated::ColumnNotProfiled),
}
}
fn column_names(side: &ReportSide) -> HashSet<&str> {
side.columns()
.iter()
.map(|column| column.name.as_str())
.collect()
}
fn baseline_reading(
observed: Option<f64>,
evidence: Evidence,
change: Option<f64>,
) -> BaselineReading {
BaselineReading {
observed: observed.map(MetricValue::Percentage),
evidence,
change: change.map(MetricValue::Percentage),
}
}
fn not_evaluated_change(mut check: Check, reason: NotEvaluated, subject: &str) -> Check {
check.message = match &reason {
NotEvaluated::NotAssessed if check.column.is_some() => {
"no values were read for this column".to_string()
}
NotEvaluated::NotAssessed => format!("{subject} was not assessed in this report"),
NotEvaluated::ColumnNotProfiled => "this column has no profile in the report".to_string(),
NotEvaluated::NotInBaseline => format!("the baseline carries no value for {subject}"),
NotEvaluated::SchemaChanged => format!(
"the two reports profiled different columns, so {subject} has a different subject \
on each side"
),
NotEvaluated::DimensionsChanged => {
format!("{subject} averages a different set of assessed dimensions on each side")
}
NotEvaluated::DefinitionNotComparable { .. } => format!(
"a measurement definition {subject} depends on is not recorded identically in both \
reports"
),
NotEvaluated::SettingChanged { .. } => format!(
"a setting that moves {subject} on unchanged data differs between the two reports"
),
NotEvaluated::SettingUnrecorded { .. } => format!(
"a setting that moves {subject} on unchanged data is not recorded in both reports"
),
NotEvaluated::QualityUnavailable { .. } => {
"the report carries no quality assessment".to_string()
}
NotEvaluated::EvidenceIncomplete { .. } => evidence_message(subject),
};
check.status = CheckStatus::NotEvaluated(reason);
check
}
fn evidence_message(subject: &str) -> String {
format!(
"{subject} was computed over part of the source on at least one side, which bounds \
nothing about the rest"
)
}
#[derive(Debug, Clone, Copy)]
enum Definition {
NullTokens,
PatternMatchCount,
RangeRuleCount,
KeyUniquenessUnits,
OutlierUnits,
KeyColumnSource,
}
impl Definition {
fn key(self) -> &'static str {
match self {
Self::NullTokens => "null_tokens",
Self::PatternMatchCount => "pattern_match_count",
Self::RangeRuleCount => "range_rule_count",
Self::KeyUniquenessUnits => "key_uniqueness_units",
Self::OutlierUnits => "outlier_units",
Self::KeyColumnSource => "key_column_source",
}
}
fn same(self, a: &MetricSemantics, b: &MetricSemantics) -> bool {
fn agree<T: PartialEq>(a: &Option<T>, b: &Option<T>) -> bool {
a.is_some() && a == b
}
match self {
Self::NullTokens => agree(&a.null_tokens, &b.null_tokens),
Self::PatternMatchCount => agree(&a.pattern_match_count, &b.pattern_match_count),
Self::RangeRuleCount => agree(&a.range_rule_count, &b.range_rule_count),
Self::KeyUniquenessUnits => agree(&a.key_uniqueness_units, &b.key_uniqueness_units),
Self::OutlierUnits => agree(&a.outlier_units, &b.outlier_units),
Self::KeyColumnSource => agree(&a.key_column_source, &b.key_column_source),
}
}
}
fn score_definitions(dimension: Option<QualityDimension>) -> &'static [Definition] {
match dimension {
None => &[
Definition::NullTokens,
Definition::PatternMatchCount,
Definition::RangeRuleCount,
Definition::KeyUniquenessUnits,
Definition::OutlierUnits,
Definition::KeyColumnSource,
],
Some(QualityDimension::Validity) => {
&[Definition::NullTokens, Definition::PatternMatchCount]
}
Some(QualityDimension::Accuracy) => &[
Definition::NullTokens,
Definition::RangeRuleCount,
Definition::OutlierUnits,
],
Some(QualityDimension::Uniqueness) => &[
Definition::NullTokens,
Definition::KeyUniquenessUnits,
Definition::KeyColumnSource,
],
Some(_) => &[Definition::NullTokens],
}
}
fn definition_gap(
a: Option<&MetricSemantics>,
b: Option<&MetricSemantics>,
definitions: &[Definition],
) -> Option<NotEvaluated> {
definitions
.iter()
.find(|definition| !a.zip(b).is_some_and(|(a, b)| definition.same(a, b)))
.map(|definition| NotEvaluated::DefinitionNotComparable {
definition: definition.key().to_string(),
})
}
type HintColumns = fn(&RecordedHints) -> &Vec<String>;
#[derive(Debug, Clone, Copy)]
enum Agreement {
Same,
Changed,
Unrecorded,
}
fn agreement<T: PartialEq + ?Sized>(a: Option<&T>, b: Option<&T>) -> Agreement {
match (a, b) {
(Some(a), Some(b)) if a == b => Agreement::Same,
(Some(_), Some(_)) => Agreement::Changed,
_ => Agreement::Unrecorded,
}
}
fn setting_gap(
configuration: &ConfigurationComparison,
identifiers: &Sides<Option<&[String]>>,
dimension: Option<QualityDimension>,
) -> Option<NotEvaluated> {
let hints = &configuration.semantic_hints;
let reads = |dimensions: &[QualityDimension]| {
dimension.is_none_or(|dimension| dimensions.contains(&dimension))
};
let hint = |columns: HintColumns| {
agreement(hints.a.as_ref().map(columns), hints.b.as_ref().map(columns))
};
let recorded = |same: Option<bool>| match same {
Some(true) => Agreement::Same,
Some(false) => Agreement::Changed,
None => Agreement::Unrecorded,
};
let pairs = match hint(|hints| &hints.temporal_pair_columns) {
Agreement::Same
if hints
.a
.as_ref()
.is_some_and(|hints| !hints.temporal_pair_columns.is_empty()) =>
{
Agreement::Unrecorded
}
other => other,
};
let settings = [
(
"score_weights",
dimension.is_none(),
recorded(configuration.score_weights.same),
),
(
"identifier_columns",
dimension != Some(QualityDimension::Completeness),
agreement(identifiers.a, identifiers.b),
),
(
"positive_columns",
reads(&[QualityDimension::Accuracy]),
hint(|hints| &hints.positive_columns),
),
(
"temporal_columns",
reads(&[QualityDimension::Timeliness]),
hint(|hints| &hints.temporal_columns),
),
(
"temporal_pairs",
reads(&[QualityDimension::Timeliness]),
pairs,
),
(
"pattern_locale",
reads(&[QualityDimension::Validity]),
recorded(configuration.pattern_locale.same),
),
];
settings
.into_iter()
.filter(|(_, read, _)| *read)
.find_map(|(setting, _, agreement)| match agreement {
Agreement::Same => None,
Agreement::Changed => Some(NotEvaluated::SettingChanged {
setting: setting.to_string(),
}),
Agreement::Unrecorded => Some(NotEvaluated::SettingUnrecorded {
setting: setting.to_string(),
}),
})
}
fn check_percentage(
code: CheckCode,
subject: Option<String>,
value: f64,
) -> Result<(), PolicyError> {
if value.is_finite() && (0.0..=PERCENTAGE_MAX).contains(&value) {
return Ok(());
}
Err(PolicyError::ThresholdOutOfRange {
code,
subject,
value,
})
}
fn required_metrics_in_order(requested: &[RequiredMetric]) -> Vec<RequiredMetric> {
let mut ordered = Vec::new();
if requested.contains(&RequiredMetric::Quality) {
ordered.push(RequiredMetric::Quality);
}
for dimension in QualityDimension::all() {
if requested.contains(&RequiredMetric::Dimension(dimension)) {
ordered.push(RequiredMetric::Dimension(dimension));
}
}
ordered
}
fn verdict_of(checks: &[Check]) -> Verdict {
if checks.iter().any(Check::is_violation) {
Verdict::Fail
} else if checks.iter().any(|check| check.not_evaluated().is_some()) {
Verdict::Inconclusive
} else {
Verdict::Pass
}
}
fn unavailable_quality(mut check: Check, report: &ProfileReport) -> Check {
let quality_status = quality_status_name(&report.quality_status).to_string();
check.message = match &report.quality_status {
QualityAnalysisStatus::NotRequested => {
"quality metrics were not requested for this run".to_string()
}
QualityAnalysisStatus::NoData => {
"quality was requested but no sample was available to measure".to_string()
}
QualityAnalysisStatus::WithheldByProjection => {
"quality was withheld: the requested dimensions measure whole rows and only \
some columns were profiled"
.to_string()
}
QualityAnalysisStatus::Failed { .. } => "the quality computation failed".to_string(),
QualityAnalysisStatus::Unrecorded | QualityAnalysisStatus::Computed => {
"the report carries no quality assessment".to_string()
}
};
check.status = CheckStatus::NotEvaluated(NotEvaluated::QualityUnavailable { quality_status });
check
}
fn aggregate_message(status: &CheckStatus, subject: &str) -> String {
match status {
CheckStatus::Passed => format!("{subject} meets the required minimum"),
CheckStatus::Failed => format!("{subject} is below the required minimum"),
CheckStatus::NotEvaluated(_) => format!(
"{subject} was computed over part of the source, which bounds nothing about \
the rest"
),
}
}
fn bounded_message(status: &CheckStatus, subject: &str) -> String {
match status {
CheckStatus::Passed => format!(
"{subject} was computed over a sample, and its whole-source interval meets the \
required minimum"
),
CheckStatus::Failed => format!(
"{subject} was computed over a sample, and its whole-source interval is below the \
required minimum"
),
CheckStatus::NotEvaluated(_) => format!(
"{subject} was computed over a sample, and the required minimum lies within its \
whole-source interval"
),
}
}
fn check_bounds(bounds: &ScoreBounds, interval: Option<ScoreInterval>) -> Option<CheckBounds> {
interval.map(|interval| CheckBounds {
lower: interval.lower,
upper: interval.upper,
confidence_level: bounds.confidence_level,
})
}
pub(crate) fn quality_status_name(status: &QualityAnalysisStatus) -> &'static str {
match status {
QualityAnalysisStatus::Computed => "computed",
QualityAnalysisStatus::NotRequested => "not_requested",
QualityAnalysisStatus::NoData => "no_data",
QualityAnalysisStatus::WithheldByProjection => "withheld_by_projection",
QualityAnalysisStatus::Failed { .. } => "failed",
QualityAnalysisStatus::Unrecorded => "unrecorded",
}
}
fn null_percentage(profile: &ColumnProfile) -> Option<f64> {
(profile.total_count > 0)
.then(|| profile.null_count as f64 / profile.total_count as f64 * PERCENTAGE_MAX)
}
pub(crate) fn scan_evidence(execution: &ExecutionMetadata) -> Evidence {
scan_coverage(
!execution.source_exhausted || execution.truncation_reason.is_some(),
execution.sampling_applied,
execution.error_count > 0,
)
}
pub(crate) fn scan_coverage(truncated: bool, sampled: bool, records_skipped: bool) -> Evidence {
let reason = if truncated {
Some(EvidenceGap::Truncated)
} else if sampled {
Some(EvidenceGap::Sampled)
} else if records_skipped {
Some(EvidenceGap::RecordsSkipped)
} else {
None
};
match reason {
Some(reason) => Evidence::Incomplete { reason },
None => Evidence::Complete,
}
}
#[derive(Debug, Clone, Copy)]
pub(crate) enum Provenance {
Overall,
Dimension(QualityDimension),
Component(&'static str),
}
pub(crate) fn dimension_components(dimension: QualityDimension) -> &'static [&'static str] {
match dimension {
QualityDimension::Completeness => &["completeness"],
QualityDimension::Consistency => &["consistency"],
QualityDimension::Uniqueness => &["key_uniqueness", "duplicate_rows"],
QualityDimension::Accuracy => &["accuracy"],
QualityDimension::Timeliness => &["timeliness"],
QualityDimension::Validity => &["validity"],
QualityDimension::Precision => &["precision"],
}
}
fn is_known(label: &str) -> bool {
QualityDimension::all()
.into_iter()
.flat_map(dimension_components)
.any(|component| *component == label)
}
fn quality_evidence(quality: Option<&QualityAssessment>, provenance: Provenance) -> Evidence {
let Some(quality) = quality else {
return Evidence::Complete;
};
sampled_component_evidence(
quality.sampled_dimensions().as_deref(),
&quality.unlisted_components(),
&quality.metrics.assessed_dimensions(),
provenance,
)
}
pub(crate) fn sampled_component_evidence(
sampled: Option<&[String]>,
unlisted: &[String],
assessed: &[QualityDimension],
provenance: Provenance,
) -> Evidence {
let Some(sampled) = sampled else {
return Evidence::Incomplete {
reason: EvidenceGap::CoverageUnrecorded,
};
};
if matches!(provenance, Provenance::Overall) && !sampled.iter().all(|label| is_known(label)) {
return Evidence::Incomplete {
reason: EvidenceGap::CoverageUnrecorded,
};
}
let reads: Vec<&str> = match provenance {
Provenance::Overall => assessed
.iter()
.copied()
.flat_map(dimension_components)
.copied()
.collect(),
Provenance::Dimension(dimension) => dimension_components(dimension).to_vec(),
Provenance::Component(component) => vec![component],
};
let any_in = |labels: &[String]| {
reads
.iter()
.any(|component| labels.iter().any(|label| label == component))
};
if any_in(unlisted) {
return Evidence::Incomplete {
reason: EvidenceGap::CoverageUnrecorded,
};
}
if any_in(sampled) {
Evidence::Incomplete {
reason: EvidenceGap::QualitySampled,
}
} else {
Evidence::Complete
}
}
#[cfg(doctest)]
struct NonExhaustiveGuards;
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use dataprof_core::{
ColumnStats, DataSource, DataType, FileFormat, QualityScoreWeights, TruncationReason,
};
use dataprof_metrics::{QualityMetrics, UniquenessMetrics};
use serde_json::{Value, json};
use super::*;
#[test]
fn every_public_type_is_non_exhaustive() {
let lines: Vec<&str> = include_str!("quality_gate.rs").lines().collect();
let marked = |i: usize| {
lines[..i]
.iter()
.rev()
.map(|l| l.trim())
.take_while(|l| l.starts_with("#[") || l.starts_with("///"))
.any(|l| l == "#[non_exhaustive]")
};
let identifier = |s: &str| -> String {
s.chars()
.take_while(|c| c.is_alphanumeric() || *c == '_')
.collect()
};
let mut declared = Vec::new();
let mut variants = Vec::new();
let mut unmarked = Vec::new();
let mut in_enum: Option<String> = None;
for (i, line) in lines.iter().enumerate() {
if let Some(enum_name) = &in_enum {
if *line == "}" {
in_enum = None;
} else if let Some(rest) = line.strip_prefix(" ")
&& rest.starts_with(|c: char| c.is_ascii_uppercase())
&& rest.contains(" {")
{
let variant = format!("{enum_name}::{}", identifier(rest));
if !marked(i) {
unmarked.push(variant.clone());
}
variants.push(variant);
}
continue;
}
let Some(rest) = line
.strip_prefix("pub enum ")
.or_else(|| line.strip_prefix("pub struct "))
else {
continue;
};
let name = identifier(rest);
if !marked(i) {
unmarked.push(name.clone());
}
if line.starts_with("pub enum ") {
in_enum = Some(name.clone());
}
declared.push(name);
}
assert!(declared.len() >= 15, "{declared:?}");
assert!(variants.len() >= 6, "{variants:?}");
assert!(unmarked.is_empty(), "not #[non_exhaustive]: {unmarked:?}");
}
use crate::ReportAssembler;
fn source() -> DataSource {
DataSource::File {
path: "drop.csv".to_string(),
format: FileFormat::Csv,
size_bytes: 256,
modified_at: None,
parquet_metadata: None,
}
}
fn column(name: &str, total: usize, nulls: usize) -> ColumnProfile {
ColumnProfile {
name: name.to_string(),
data_type: DataType::String,
null_count: nulls,
total_count: total,
unique_count: None,
unique_count_is_approximate: None,
unique_count_lower_bound: None,
invalid_count: None,
type_homogeneity: None,
locale_number_count: None,
stats: ColumnStats::None,
patterns: None,
}
}
fn sample_data() -> HashMap<String, Vec<String>> {
let ids = (1..=5).map(|n| n.to_string()).collect::<Vec<_>>();
let notes = vec!["a", "b", "c", "d", ""]
.into_iter()
.map(str::to_string)
.collect::<Vec<_>>();
HashMap::from([("id".to_string(), ids), ("note".to_string(), notes)])
}
fn complete_report() -> ProfileReport {
ReportAssembler::new(source(), ExecutionMetadata::new(5, 2, 10))
.columns(vec![column("id", 5, 0), column("note", 5, 1)])
.with_quality_data(sample_data())
.build()
}
fn truncated_report() -> ProfileReport {
ReportAssembler::new(
source(),
ExecutionMetadata::new(5, 2, 10).with_truncation(TruncationReason::MaxRows(5)),
)
.columns(vec![column("id", 5, 0), column("note", 5, 1)])
.with_quality_data(sample_data())
.build()
}
fn no_quality_report() -> ProfileReport {
ReportAssembler::new(source(), ExecutionMetadata::new(5, 2, 10))
.columns(vec![column("id", 5, 0), column("note", 5, 1)])
.with_quality_data(sample_data())
.skip_quality()
.build()
}
fn duplicate_report(duplicates: usize, approximate: bool) -> ProfileReport {
let metrics = QualityMetrics {
uniqueness: Some(UniquenessMetrics {
duplicate_rows: duplicates,
key_uniqueness: 100.0,
high_cardinality_warning: false,
rows_checked: 5,
key_column: None,
key_values_checked: None,
duplicate_rows_approximate: approximate,
}),
..QualityMetrics::default()
};
ProfileReport::new(
source(),
vec![column("id", 5, 0)],
ExecutionMetadata::new(5, 1, 10).with_truncation(TruncationReason::MaxRows(5)),
Some(QualityAssessment::exact(metrics)),
)
}
fn check_for(result: &GateResult, code: CheckCode) -> &Check {
result
.checks
.iter()
.find(|check| check.code == code)
.unwrap_or_else(|| panic!("no {code} check in {:?}", result.checks))
}
fn column_check<'a>(result: &'a GateResult, name: &str) -> &'a Check {
result
.checks
.iter()
.find(|check| check.column.as_deref() == Some(name))
.unwrap_or_else(|| panic!("no check for column {name}"))
}
fn reservoir_report() -> ProfileReport {
ReportAssembler::new(source(), ExecutionMetadata::new(100, 2, 10))
.columns(vec![column("id", 100, 0), column("note", 100, 20)])
.with_quality_data(sample_data())
.build()
}
#[test]
fn a_reservoir_backed_score_is_decided_on_its_whole_source_interval() {
let report = reservoir_report();
assert!(report.execution.source_exhausted);
assert!(!report.execution.sampling_applied);
let quality = report.quality.as_ref().expect("quality was computed");
let sampled = quality
.sampled_dimensions()
.expect("a profiling run records its provenance");
assert!(
!sampled.is_empty(),
"the assembler did not bifurcate; this test no longer reaches the \
case it guards"
);
let interval = quality
.score_bounds()
.expect("a sampled assessment carries bounds")
.overall_score
.expect("every sampled dimension here is bounded");
assert!(
interval.lower < interval.upper && interval.upper < 100.0,
"a five-value sample leaves room on both sides: {interval:?}"
);
let decide = |min: f64| {
let result = QualityPolicy::new()
.min_quality_score(min)
.evaluate(&report)
.unwrap();
assert_eq!(result.evidence, Evidence::Complete);
let check = check_for(&result, CheckCode::MinQualityScore).clone();
assert_eq!(
check.evidence,
Evidence::Incomplete {
reason: EvidenceGap::QualitySampled
}
);
assert_eq!(
check.bounds.map(|bounds| (bounds.lower, bounds.upper)),
Some((interval.lower, interval.upper))
);
(result.verdict, check)
};
let (verdict, _) = decide(interval.lower);
assert_eq!(verdict, Verdict::Pass);
let (verdict, check) = decide((interval.lower + interval.upper) / 2.0);
assert_eq!(verdict, Verdict::Inconclusive);
assert_eq!(
check.not_evaluated(),
Some(&NotEvaluated::EvidenceIncomplete {
gap: EvidenceGap::QualitySampled
})
);
let (verdict, _) = decide(interval.upper + 0.01);
assert_eq!(verdict, Verdict::Fail);
}
#[test]
fn a_threshold_at_the_shown_value_passes_live_and_after_a_round_trip() {
let thirds = ReportAssembler::new(source(), ExecutionMetadata::new(3, 1, 10))
.columns(vec![column("x", 3, 1)])
.with_quality_data(HashMap::from([(
"x".to_string(),
vec!["1".to_string(), "2".to_string(), String::new()],
)]))
.build();
let mut wrong = Vec::new();
for (label, report) in [
("thirds", thirds),
("reservoir", reservoir_report()),
("complete", complete_report()),
] {
let loaded: ProfileReport =
serde_json::from_str(&serde_json::to_string(&report).unwrap()).unwrap();
let quality = report.quality.as_ref().expect("quality was computed");
let mut policies = Vec::new();
if quality
.score_bounds()
.and_then(|bounds| bounds.overall_score)
.is_none()
{
policies.push((
"overall".to_string(),
QualityPolicy::new()
.min_quality_score(rounded_2(report.quality_score().unwrap())),
));
}
for (name, score) in &quality.scores().dimension_scores {
let Some(score) = score else { continue };
let shown = quality
.score_bounds()
.and_then(|bounds| bounds.dimension_scores.get(name).copied().flatten())
.map_or(*score, |interval| interval.lower);
let dimension: QualityDimension = name.parse().unwrap();
policies.push((
name.clone(),
QualityPolicy::new().min_dimension_score(dimension, rounded_2(shown)),
));
}
for column in &report.column_profiles {
let shown = rounded_2(null_percentage(column).unwrap());
policies.push((
format!("null {}", column.name),
QualityPolicy::new().max_null_percentage(column.name.clone(), shown),
));
}
if let Some(interval) = quality
.score_bounds()
.and_then(|bounds| bounds.overall_score)
{
policies.push((
"overall lower bound".to_string(),
QualityPolicy::new().min_quality_score(rounded_2(interval.lower)),
));
}
if label == "thirds" {
policies.push((
"completeness at a third decimal".to_string(),
QualityPolicy::new()
.min_dimension_score(QualityDimension::Completeness, 66.674),
));
}
for (rule, policy) in policies {
let live = policy.evaluate(&report).unwrap().verdict;
let reloaded = policy.evaluate(&loaded).unwrap().verdict;
if live != Verdict::Pass || reloaded != Verdict::Pass {
wrong.push(format!(
"{label} {rule}: live {live:?}, loaded {reloaded:?}"
));
}
}
}
assert!(wrong.is_empty(), "{wrong:#?}");
}
#[test]
fn a_bound_that_rounds_up_decides_the_same_live_and_loaded() {
let mut report = reservoir_report();
let quality = report.quality.clone().expect("quality was computed");
let mut bounds = quality
.score_bounds()
.expect("a sampled report is bounded")
.clone();
let interval = bounds
.overall_score
.as_mut()
.expect("the overall score is bounded");
interval.lower = rounded_2(interval.lower) - 0.004;
interval.upper = rounded_2(interval.upper) - 0.004;
let (lower, upper) = (interval.lower, interval.upper);
assert!(rounded_2(lower) > lower && rounded_2(upper) > upper && lower < upper);
report.quality = Some(quality.with_score_bounds(Some(bounds)));
let loaded: ProfileReport =
serde_json::from_str(&serde_json::to_string(&report).unwrap()).unwrap();
let mut wrong = Vec::new();
for (label, min) in [("lower", rounded_2(lower)), ("upper", rounded_2(upper))] {
let policy = QualityPolicy::new().min_quality_score(min);
let live = policy.evaluate(&report).unwrap().verdict;
let reloaded = policy.evaluate(&loaded).unwrap().verdict;
if live != reloaded {
wrong.push(format!(
"min at the shown {label} bound {min}: live {live:?}, loaded {reloaded:?}"
));
}
}
assert!(wrong.is_empty(), "{wrong:#?}");
}
#[test]
fn an_observed_scope_decides_on_the_sampled_score_itself() {
let report = reservoir_report();
let score = report.quality_score().unwrap();
let result = QualityPolicy::new()
.min_quality_score(score)
.scope(PolicyScope::Observed)
.evaluate(&report)
.unwrap();
let check = check_for(&result, CheckCode::MinQualityScore);
assert_eq!(result.verdict, Verdict::Pass);
assert_eq!(check.bounds, None);
}
#[test]
fn a_sampled_score_without_recorded_bounds_stays_unevaluated() {
let mut report = reservoir_report();
let quality = report.quality.take().unwrap();
report.quality = Some(QualityAssessment::new(quality.metrics, quality.confidence));
let result = QualityPolicy::new()
.min_quality_score(1.0)
.evaluate(&report)
.unwrap();
assert_eq!(result.verdict, Verdict::Inconclusive);
assert_eq!(check_for(&result, CheckCode::MinQualityScore).bounds, None);
}
#[test]
fn an_exactly_counted_dimension_survives_another_dimensions_sampling() {
let report = reservoir_report();
let sampled = report
.quality
.as_ref()
.expect("quality was computed")
.sampled_dimensions()
.expect("a profiling run records its provenance");
assert!(
!sampled.iter().any(|label| label == "completeness"),
"completeness is supposed to come from exact column counters"
);
let result = QualityPolicy::new()
.min_dimension_score(QualityDimension::Completeness, 1.0)
.min_dimension_score(QualityDimension::Consistency, 1.0)
.evaluate(&report)
.unwrap();
assert_eq!(
result
.checks
.iter()
.find(|check| check.dimension == Some(QualityDimension::Completeness))
.map(|check| check.evidence),
Some(Evidence::Complete)
);
assert_eq!(
result
.checks
.iter()
.find(|check| check.dimension == Some(QualityDimension::Consistency))
.map(|check| check.evidence),
Some(Evidence::Incomplete {
reason: EvidenceGap::QualitySampled
})
);
}
#[test]
fn a_legacy_document_without_recorded_coverage_is_not_read_as_a_full_scan() {
let document = serde_json::to_value(complete_report()).expect("serializes");
let mut legacy = document.clone();
legacy["quality"] = document["quality"]["metrics"].clone();
let report: ProfileReport = serde_json::from_value(legacy).expect("legacy document loads");
assert_eq!(
report
.quality
.as_ref()
.expect("the metrics survived")
.sampled_dimensions(),
None
);
let result = QualityPolicy::new()
.min_quality_score(1.0)
.evaluate(&report)
.unwrap();
assert_eq!(result.verdict, Verdict::Inconclusive);
let check = check_for(&result, CheckCode::MinQualityScore);
assert_eq!(
check.evidence,
Evidence::Incomplete {
reason: EvidenceGap::CoverageUnrecorded
}
);
let observed = QualityPolicy::new()
.min_quality_score(1.0)
.scope(PolicyScope::Observed)
.evaluate(&report)
.unwrap();
assert_eq!(observed.verdict, Verdict::Pass);
}
#[test]
fn a_zero_weighted_sampled_dimension_does_not_taint_the_overall_score() {
let mut report = reservoir_report();
let quality = report.quality.as_mut().expect("quality was computed");
let sampled = quality
.sampled_dimensions()
.expect("a profiling run records its provenance");
assert!(
sampled.iter().any(|label| label == "consistency"),
"consistency is supposed to come from the reservoir"
);
quality.metrics.score_weights = QualityScoreWeights {
completeness: 1.0,
consistency: 0.0,
uniqueness: 0.0,
accuracy: 0.0,
timeliness: 0.0,
validity: 0.0,
precision: 0.0,
};
let result = QualityPolicy::new()
.min_quality_score(1.0)
.min_dimension_score(QualityDimension::Consistency, 1.0)
.evaluate(&report)
.unwrap();
assert_eq!(
check_for(&result, CheckCode::MinQualityScore).evidence,
Evidence::Complete
);
assert_eq!(
check_for(&result, CheckCode::MinDimensionScore).evidence,
Evidence::Incomplete {
reason: EvidenceGap::QualitySampled
}
);
}
#[test]
fn zero_weights_do_not_erase_the_sampling_record() {
let mut report = reservoir_report();
let quality = report.quality.as_mut().expect("quality was computed");
quality.metrics.score_weights = QualityScoreWeights {
completeness: 0.0,
consistency: 0.0,
uniqueness: 0.0,
accuracy: 0.0,
timeliness: 0.0,
validity: 0.0,
precision: 0.0,
};
let reassessed =
QualityAssessment::new(quality.metrics.clone(), quality.confidence.clone());
assert!(reassessed.metrics.assessed_dimensions().is_empty());
assert!(
reassessed
.sampled_dimensions()
.expect("provenance survives")
.iter()
.any(|label| label == "consistency"),
"a measured dimension kept no record of coming from a sample"
);
}
#[test]
fn met_requirements_pass() {
let result = QualityPolicy::new()
.min_quality_score(1.0)
.max_null_percentage_any(50.0)
.evaluate(&complete_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Pass);
assert!(result.passed());
assert_eq!(result.violations().count(), 0);
assert_eq!(result.unevaluated().count(), 0);
}
#[test]
fn violated_requirement_fails() {
let result = QualityPolicy::new()
.max_null_percentage("note", 10.0)
.evaluate(&complete_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Fail);
assert!(!result.passed());
let violation = result.violations().next().expect("one violation");
assert_eq!(violation.code, CheckCode::MaxNullPercentage);
assert_eq!(violation.column.as_deref(), Some("note"));
assert_eq!(violation.observed, Some(MetricValue::Percentage(20.0)));
assert_eq!(
violation.expected,
Expectation::AtMost {
value: MetricValue::Percentage(10.0)
}
);
}
#[test]
fn named_limit_overrides_the_wildcard() {
let result = QualityPolicy::new()
.max_null_percentage("note", 50.0)
.max_null_percentage_any(0.0)
.evaluate(&complete_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Pass);
assert_eq!(result.checks.len(), 2);
assert_eq!(
column_check(&result, "note").expected,
Expectation::AtMost {
value: MetricValue::Percentage(50.0)
}
);
assert_eq!(
column_check(&result, "id").expected,
Expectation::AtMost {
value: MetricValue::Percentage(0.0)
}
);
}
#[test]
fn thresholds_are_on_the_percentage_scale() {
let result = QualityPolicy::new()
.min_quality_score(0.9)
.evaluate(&complete_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Pass);
let err = QualityPolicy::new()
.min_quality_score(101.0)
.evaluate(&complete_report())
.unwrap_err();
assert_eq!(
err,
PolicyError::ThresholdOutOfRange {
code: CheckCode::MinQualityScore,
subject: None,
value: 101.0,
}
);
assert!(
QualityPolicy::new()
.max_null_percentage("id", f64::NAN)
.validate()
.is_err()
);
}
#[test]
fn empty_policy_is_rejected_rather_than_passing_everything() {
assert_eq!(
QualityPolicy::new()
.evaluate(&complete_report())
.unwrap_err(),
PolicyError::NoRequirements
);
}
#[test]
fn unanalyzed_quality_is_inconclusive_not_a_pass() {
let result = QualityPolicy::new()
.min_quality_score(90.0)
.evaluate(&no_quality_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Inconclusive);
assert!(!result.passed());
let check = check_for(&result, CheckCode::MinQualityScore);
assert_eq!(check.observed, None);
assert_eq!(
check.not_evaluated(),
Some(&NotEvaluated::QualityUnavailable {
quality_status: "not_requested".to_string(),
})
);
}
#[test]
fn require_quality_turns_absence_into_a_violation() {
let result = QualityPolicy::new()
.require_quality()
.min_quality_score(90.0)
.evaluate(&no_quality_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Fail);
let required = check_for(&result, CheckCode::RequireMetric);
assert!(required.is_violation());
assert_eq!(required.expected, Expectation::Analyzed);
assert_eq!(required.observed, None);
}
#[test]
fn require_dimension_fails_when_the_dimension_had_nothing_to_assess() {
let result = QualityPolicy::new()
.require_dimension(QualityDimension::Timeliness)
.evaluate(&complete_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Fail);
assert_eq!(
check_for(&result, CheckCode::RequireMetric).dimension,
Some(QualityDimension::Timeliness)
);
}
#[test]
fn unassessed_dimension_is_distinct_from_unrequested_quality() {
let result = QualityPolicy::new()
.min_dimension_score(QualityDimension::Timeliness, 90.0)
.evaluate(&complete_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Inconclusive);
assert_eq!(
check_for(&result, CheckCode::MinDimensionScore).not_evaluated(),
Some(&NotEvaluated::NotAssessed)
);
}
#[test]
fn unprofiled_column_is_not_evaluated() {
let result = QualityPolicy::new()
.max_null_percentage("absent", 0.0)
.evaluate(&complete_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Inconclusive);
assert_eq!(
column_check(&result, "absent").not_evaluated(),
Some(&NotEvaluated::ColumnNotProfiled)
);
}
#[test]
fn a_ratio_over_a_partial_scan_cannot_pass_a_full_source_policy() {
let report = truncated_report();
let result = QualityPolicy::new()
.max_null_percentage_any(50.0)
.evaluate(&report)
.unwrap();
assert_eq!(result.verdict, Verdict::Inconclusive);
assert_eq!(
result.evidence,
Evidence::Incomplete {
reason: EvidenceGap::Truncated
}
);
let check = column_check(&result, "note");
assert_eq!(
check.not_evaluated(),
Some(&NotEvaluated::EvidenceIncomplete {
gap: EvidenceGap::Truncated
})
);
assert_eq!(check.observed, Some(MetricValue::Percentage(20.0)));
let observed = QualityPolicy::new()
.max_null_percentage_any(50.0)
.scope(PolicyScope::Observed)
.evaluate(&report)
.unwrap();
assert_eq!(observed.verdict, Verdict::Pass);
}
#[test]
fn a_witnessed_duplicate_fails_despite_a_partial_scan() {
let result = QualityPolicy::new()
.max_duplicate_rows(0)
.evaluate(&duplicate_report(2, false))
.unwrap();
assert_eq!(result.verdict, Verdict::Fail);
let check = check_for(&result, CheckCode::MaxDuplicateRows);
assert_eq!(check.observed, Some(MetricValue::Count(2)));
assert_eq!(
check.evidence,
Evidence::Incomplete {
reason: EvidenceGap::Truncated
}
);
}
#[test]
fn no_duplicate_in_a_partial_scan_is_not_a_pass() {
let result = QualityPolicy::new()
.max_duplicate_rows(0)
.evaluate(&duplicate_report(0, false))
.unwrap();
assert_eq!(result.verdict, Verdict::Inconclusive);
assert_eq!(
check_for(&result, CheckCode::MaxDuplicateRows).not_evaluated(),
Some(&NotEvaluated::EvidenceIncomplete {
gap: EvidenceGap::Truncated
})
);
}
#[test]
fn an_estimated_duplicate_count_settles_neither_direction() {
let result = QualityPolicy::new()
.max_duplicate_rows(0)
.evaluate(&duplicate_report(2, true))
.unwrap();
assert_eq!(result.verdict, Verdict::Inconclusive);
let check = check_for(&result, CheckCode::MaxDuplicateRows);
assert_eq!(
check.not_evaluated(),
Some(&NotEvaluated::EvidenceIncomplete {
gap: EvidenceGap::Truncated
})
);
assert_eq!(
check.message,
"the duplicate-row count is an estimate, which can neither witness \
duplicates above the allowance nor rule them out"
);
}
#[test]
fn a_null_count_ceiling_is_decided_on_the_exact_count() {
let report = ReportAssembler::new(source(), ExecutionMetadata::new(20_001, 2, 10))
.columns(vec![column("key", 20_001, 1), column("clean", 20_001, 0)])
.build();
let percentage = QualityPolicy::new()
.max_null_percentage("key", 0.0)
.evaluate(&report)
.unwrap();
assert_eq!(percentage.verdict, Verdict::Pass);
let count = QualityPolicy::new()
.max_null_count("key", 0)
.max_null_count_any(0)
.evaluate(&report)
.unwrap();
assert_eq!(count.verdict, Verdict::Fail);
let key = column_check(&count, "key");
assert_eq!(key.code, CheckCode::MaxNullCount);
assert_eq!(key.observed, Some(MetricValue::Count(1)));
assert_eq!(key.status, CheckStatus::Failed);
assert_eq!(column_check(&count, "clean").status, CheckStatus::Passed);
}
#[test]
fn witnessed_nulls_fail_a_partial_scan_and_a_clean_prefix_does_not_pass() {
let result = QualityPolicy::new()
.max_null_count_any(0)
.evaluate(&truncated_report())
.unwrap();
assert_eq!(result.verdict, Verdict::Fail);
assert_eq!(column_check(&result, "note").status, CheckStatus::Failed);
assert_eq!(
column_check(&result, "id").not_evaluated(),
Some(&NotEvaluated::EvidenceIncomplete {
gap: EvidenceGap::Truncated
})
);
let observed = QualityPolicy::new()
.max_null_count("note", 1)
.scope(PolicyScope::Observed)
.evaluate(&truncated_report())
.unwrap();
assert_eq!(observed.verdict, Verdict::Pass);
}
#[test]
fn a_column_with_no_values_read_has_a_null_count_of_zero() {
let report = ReportAssembler::new(source(), ExecutionMetadata::new(0, 1, 10))
.columns(vec![column("key", 0, 0)])
.build();
let count = QualityPolicy::new()
.max_null_count("key", 0)
.evaluate(&report)
.unwrap();
assert_eq!(count.verdict, Verdict::Pass);
assert_eq!(
column_check(&count, "key").observed,
Some(MetricValue::Count(0))
);
let percentage = QualityPolicy::new()
.max_null_percentage("key", 0.0)
.evaluate(&report)
.unwrap();
assert_eq!(
column_check(&percentage, "key").not_evaluated(),
Some(&NotEvaluated::NotAssessed)
);
}
#[test]
fn a_named_null_count_overrides_the_wildcard_and_absence_is_unevaluated() {
let result = QualityPolicy::new()
.max_null_count("note", 1)
.max_null_count("absent", 0)
.max_null_count_any(0)
.evaluate(&complete_report())
.unwrap();
let checks: Vec<_> = result
.checks
.iter()
.map(|check| (check.column.clone().unwrap(), check.expected))
.collect();
assert_eq!(
checks,
vec![
(
"absent".to_string(),
Expectation::AtMost {
value: MetricValue::Count(0)
}
),
(
"note".to_string(),
Expectation::AtMost {
value: MetricValue::Count(1)
}
),
(
"id".to_string(),
Expectation::AtMost {
value: MetricValue::Count(0)
}
),
]
);
assert_eq!(
column_check(&result, "absent").not_evaluated(),
Some(&NotEvaluated::ColumnNotProfiled)
);
assert_eq!(column_check(&result, "note").status, CheckStatus::Passed);
assert_eq!(result.verdict, Verdict::Inconclusive);
}
#[test]
fn a_violation_outranks_an_unevaluated_check() {
let result = QualityPolicy::new()
.max_duplicate_rows(0)
.min_dimension_score(QualityDimension::Timeliness, 90.0)
.evaluate(&duplicate_report(2, false))
.unwrap();
assert_eq!(result.verdict, Verdict::Fail);
assert_eq!(result.unevaluated().count(), 1);
}
#[test]
fn evaluation_order_does_not_depend_on_insertion_order() {
let report = complete_report();
let forward = QualityPolicy::new()
.max_null_percentage("note", 50.0)
.max_null_percentage("id", 50.0)
.evaluate(&report)
.unwrap();
let reverse = QualityPolicy::new()
.max_null_percentage("id", 50.0)
.max_null_percentage("note", 50.0)
.evaluate(&report)
.unwrap();
assert_eq!(forward, reverse);
let columns: Vec<_> = forward
.checks
.iter()
.map(|check| check.column.clone().unwrap())
.collect();
assert_eq!(columns, vec!["id".to_string(), "note".to_string()]);
}
#[test]
fn result_serializes_into_the_documented_shape() {
let result = QualityPolicy::new()
.max_null_percentage("note", 10.0)
.evaluate(&complete_report())
.unwrap();
assert_eq!(
serde_json::to_value(&result).unwrap(),
json!({
"verdict": "fail",
"scope": "full_source",
"evidence": {"coverage": "complete"},
"checks": [{
"code": "max_null_percentage",
"column": "note",
"expected": {"comparison": "at_most", "value": 10.0},
"observed": 20.0,
"scope": "full_source",
"evidence": {"coverage": "complete"},
"status": "failed",
"message": "this column's null percentage is above the allowance",
}],
})
);
}
#[test]
fn an_unevaluated_check_serializes_its_reason() {
let result = QualityPolicy::new()
.min_quality_score(90.0)
.evaluate(&no_quality_report())
.unwrap();
let check = &serde_json::to_value(&result).unwrap()["checks"][0];
assert_eq!(check["status"], Value::String("not_evaluated".to_string()));
assert_eq!(
check["reason"],
Value::String("quality_unavailable".to_string())
);
assert_eq!(
check["quality_status"],
Value::String("not_requested".to_string())
);
assert!(check.get("observed").is_none());
}
}