1use std::collections::{BTreeMap, HashSet};
62
63use dataprof_core::serde_helpers::rounded_2;
64use dataprof_core::{
65 DataSource, DataType, FileFormat, QualityDimension, QualityScoreWeights, SemanticHintBinding,
66 SemanticHintKind,
67};
68use serde::ser::{SerializeMap, Serializer};
69
70use crate::profile_report::{
71 AnalysisSettings, ColumnSelection, MetricSemantics, PatternLocale, ProfileReport,
72};
73use crate::quality_gate::{
74 Evidence, EvidenceGap, Provenance, sampled_component_evidence, scan_coverage, scan_evidence,
75};
76
77#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
79#[non_exhaustive]
80pub struct Sides<T> {
81 pub a: T,
83 pub b: T,
85}
86
87#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
89#[non_exhaustive]
90pub struct ReportReference {
91 pub id: Option<String>,
94 pub timestamp: Option<String>,
96 pub source: String,
98 pub rows_processed: Option<usize>,
100 pub evidence: Evidence,
102}
103
104#[derive(Debug, Clone, PartialEq, serde::Serialize)]
106#[non_exhaustive]
107pub struct ScoreChange {
108 pub a: Option<f64>,
110 pub b: Option<f64>,
112 pub abs: Option<f64>,
114 pub rel_pct: Option<f64>,
117 pub evidence: Sides<Evidence>,
120}
121
122impl ScoreChange {
123 fn new(a: Option<f64>, b: Option<f64>, evidence: Sides<Evidence>) -> Self {
124 let (abs, rel_pct) = match (a, b) {
125 (Some(a), Some(b)) => (
126 Some(rounded_2(b - a)),
127 (a != 0.0).then(|| rounded_2((b - a) / a.abs() * 100.0)),
128 ),
129 _ => (None, None),
130 };
131 Self {
132 a,
133 b,
134 abs,
135 rel_pct,
136 evidence,
137 }
138 }
139}
140
141#[derive(Debug, Clone, PartialEq)]
144#[non_exhaustive]
145pub struct DimensionChange {
146 pub dimension: QualityDimension,
148 pub score: ScoreChange,
150}
151
152#[derive(Debug, Clone, PartialEq, serde::Serialize)]
156#[non_exhaustive]
157pub struct ColumnChange {
158 #[serde(skip)]
160 pub name: String,
161 pub null_pct_a: Option<f64>,
164 pub null_pct_b: Option<f64>,
166 pub null_pct_delta: Option<f64>,
169}
170
171#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
173#[non_exhaustive]
174pub struct SemanticsComparison {
175 pub a: Option<MetricSemantics>,
177 pub b: Option<MetricSemantics>,
179 pub comparable: Option<bool>,
184}
185
186#[derive(Debug, Clone, PartialEq, serde::Serialize)]
188#[non_exhaustive]
189pub struct SettingComparison<T> {
190 pub a: Option<T>,
192 pub b: Option<T>,
194 pub same: Option<bool>,
198}
199
200impl<T: PartialEq> SettingComparison<T> {
201 fn new(a: Option<T>, b: Option<T>) -> Self {
202 let same = match (&a, &b) {
203 (Some(left), Some(right)) => Some(left == right),
204 _ => None,
205 };
206 Self { a, b, same }
207 }
208}
209
210#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
219#[non_exhaustive]
220pub struct RecordedHints {
221 pub positive_columns: Vec<String>,
223 pub identifier_columns: Vec<String>,
225 pub temporal_columns: Vec<String>,
227 pub temporal_pair_columns: Vec<String>,
230}
231
232impl RecordedHints {
233 fn from_bindings(bindings: &[SemanticHintBinding]) -> Self {
234 let columns = |kind: SemanticHintKind| {
235 let mut names: Vec<String> = bindings
236 .iter()
237 .filter(|binding| binding.kind == kind)
238 .map(|binding| binding.column.clone())
239 .collect();
240 names.sort();
241 names
242 };
243 Self {
244 positive_columns: columns(SemanticHintKind::Positive),
245 identifier_columns: columns(SemanticHintKind::Identifier),
246 temporal_columns: columns(SemanticHintKind::Temporal),
247 temporal_pair_columns: columns(SemanticHintKind::TemporalPair),
248 }
249 }
250}
251
252#[derive(Debug, Clone, PartialEq, serde::Serialize)]
254#[non_exhaustive]
255pub struct ConfigurationComparison {
256 pub score_weights: SettingComparison<QualityScoreWeights>,
260 pub semantic_hints: SettingComparison<RecordedHints>,
263 pub pattern_locale: SettingComparison<PatternLocale>,
266}
267
268#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
270#[non_exhaustive]
271pub struct SchemaChange {
272 pub added: Vec<String>,
274 pub removed: Vec<String>,
276 pub common: Vec<String>,
278}
279
280#[derive(Debug, Clone, PartialEq, serde::Serialize)]
283#[non_exhaustive]
284pub struct ReportDelta {
285 pub quality_score: ScoreChange,
287 #[serde(serialize_with = "serialize_dimensions")]
290 pub dimensions: Vec<DimensionChange>,
291 #[serde(serialize_with = "serialize_columns")]
294 pub columns: Vec<ColumnChange>,
295 pub metric_semantics: SemanticsComparison,
297 pub configuration: ConfigurationComparison,
299 pub schema: SchemaChange,
301 pub reports: Sides<ReportReference>,
303}
304
305impl ReportDelta {
306 pub fn between(a: &ProfileReport, b: &ProfileReport) -> Self {
308 Self::between_sides(&ReportSide::from_report(a), &ReportSide::from_report(b))
309 }
310
311 pub fn between_sides(a: &ReportSide, b: &ReportSide) -> Self {
313 let pair = |a: Evidence, b: Evidence| Sides { a, b };
314 let quality_score = ScoreChange::new(
315 a.quality_score,
316 b.quality_score,
317 pair(a.overall_evidence, b.overall_evidence),
318 );
319 let dimensions = a
320 .dimensions
321 .iter()
322 .zip(&b.dimensions)
323 .map(|(left, right)| {
324 debug_assert_eq!(left.dimension, right.dimension);
325 DimensionChange {
326 dimension: left.dimension,
327 score: ScoreChange::new(
328 left.score,
329 right.score,
330 pair(left.evidence, right.evidence),
331 ),
332 }
333 })
334 .collect();
335
336 let a_names: HashSet<&str> = a
337 .columns
338 .iter()
339 .map(|column| column.name.as_str())
340 .collect();
341 let b_names: HashSet<&str> = b
342 .columns
343 .iter()
344 .map(|column| column.name.as_str())
345 .collect();
346 let a_nulls: BTreeMap<&str, Option<f64>> = a
347 .columns
348 .iter()
349 .map(|column| (column.name.as_str(), column.null_pct))
350 .collect();
351 let b_nulls: BTreeMap<&str, Option<f64>> = b
352 .columns
353 .iter()
354 .map(|column| (column.name.as_str(), column.null_pct))
355 .collect();
356 let added: Vec<String> = b
357 .columns
358 .iter()
359 .filter(|column| !a_names.contains(column.name.as_str()))
360 .map(|column| column.name.clone())
361 .collect();
362 let columns = a
363 .columns
364 .iter()
365 .map(|column| &column.name)
366 .chain(&added)
367 .map(|name| {
368 let null_pct_a = a_nulls.get(name.as_str()).copied().flatten();
369 let null_pct_b = b_nulls.get(name.as_str()).copied().flatten();
370 ColumnChange {
371 name: name.clone(),
372 null_pct_a,
373 null_pct_b,
374 null_pct_delta: null_pct_a.zip(null_pct_b).map(|(a, b)| rounded_2(b - a)),
375 }
376 })
377 .collect();
378 let (removed, common) = a
379 .columns
380 .iter()
381 .map(|column| column.name.clone())
382 .partition(|name| !b_names.contains(name.as_str()));
383
384 let comparable = match (&a.metric_semantics, &b.metric_semantics) {
385 (Some(left), Some(right)) if left.is_fully_recorded() && right.is_fully_recorded() => {
386 Some(left == right)
387 }
388 _ => None,
389 };
390
391 Self {
392 quality_score,
393 dimensions,
394 columns,
395 metric_semantics: SemanticsComparison {
396 a: a.metric_semantics.clone(),
397 b: b.metric_semantics.clone(),
398 comparable,
399 },
400 configuration: ConfigurationComparison {
401 score_weights: SettingComparison::new(a.score_weights, b.score_weights),
402 semantic_hints: SettingComparison::new(
403 a.semantic_hints.clone(),
404 b.semantic_hints.clone(),
405 ),
406 pattern_locale: SettingComparison::new(a.pattern_locale, b.pattern_locale),
407 },
408 schema: SchemaChange {
409 added,
410 removed,
411 common,
412 },
413 reports: Sides {
414 a: a.reference.clone(),
415 b: b.reference.clone(),
416 },
417 }
418 }
419}
420
421fn serialize_dimensions<S: Serializer>(
422 dimensions: &[DimensionChange],
423 serializer: S,
424) -> Result<S::Ok, S::Error> {
425 let mut map = serializer.serialize_map(Some(dimensions.len()))?;
426 for change in dimensions {
427 map.serialize_entry(&change.dimension.to_string(), &change.score)?;
428 }
429 map.end()
430}
431
432fn serialize_columns<S: Serializer>(
433 columns: &[ColumnChange],
434 serializer: S,
435) -> Result<S::Ok, S::Error> {
436 let mut map = serializer.serialize_map(Some(columns.len()))?;
437 for change in columns {
438 map.serialize_entry(&change.name, change)?;
439 }
440 map.end()
441}
442
443#[derive(Debug, Clone, PartialEq)]
452#[non_exhaustive]
453pub struct ReportSide {
454 reference: ReportReference,
455 quality_score: Option<f64>,
456 overall_evidence: Evidence,
457 dimensions: Vec<DimensionSide>,
458 columns: Vec<ColumnSide>,
459 metric_semantics: Option<MetricSemantics>,
460 score_weights: Option<QualityScoreWeights>,
461 semantic_hints: Option<RecordedHints>,
462 identifier_order: Option<Vec<String>>,
466 pattern_locale: Option<PatternLocale>,
467 column_selection: Option<ColumnSelection>,
469 schema_declared: bool,
472}
473
474fn identifier_order(bindings: &[SemanticHintBinding]) -> Vec<String> {
476 bindings
477 .iter()
478 .filter(|binding| binding.kind == SemanticHintKind::Identifier)
479 .map(|binding| binding.column.clone())
480 .collect()
481}
482
483#[derive(Debug, Clone, PartialEq)]
484struct DimensionSide {
485 dimension: QualityDimension,
486 score: Option<f64>,
487 evidence: Evidence,
488}
489
490#[derive(Debug, Clone, PartialEq)]
492pub(crate) struct ColumnSide {
493 pub(crate) name: String,
494 pub(crate) null_pct: Option<f64>,
497 pub(crate) data_type: Option<DataType>,
500 pub(crate) counts: Option<(usize, usize)>,
504}
505
506impl ColumnSide {
507 pub(crate) fn populated(&self) -> Option<bool> {
510 self.counts.map(|(total, nulls)| nulls < total)
511 }
512
513 pub(crate) fn all_null(&self) -> Option<bool> {
516 self.counts
517 .map(|(total, nulls)| total > 0 && nulls == total)
518 }
519}
520
521fn first_of_each_name(columns: impl Iterator<Item = ColumnSide>) -> Vec<ColumnSide> {
526 let mut seen = HashSet::new();
527 columns
528 .filter(|column| seen.insert(column.name.clone()))
529 .collect()
530}
531
532fn at_precision(value: Option<f64>) -> Option<f64> {
535 value.filter(|value| value.is_finite()).map(rounded_2)
536}
537
538impl ReportSide {
539 pub fn from_report(report: &ProfileReport) -> Self {
541 let scan = scan_evidence(&report.execution);
542 let quality = report.quality.as_ref();
543 let scores = quality.map(|assessment| assessment.scores());
544 let sampled = quality.map(|assessment| assessment.sampled_dimensions());
545 let assessed = quality
546 .map(|assessment| assessment.metrics.assessed_dimensions())
547 .unwrap_or_default();
548 let unlisted = quality
549 .map(|assessment| assessment.unlisted_components())
550 .unwrap_or_default();
551 let evidence = |provenance| match &sampled {
554 Some(sampled) => scan.and(sampled_component_evidence(
555 sampled.as_deref(),
556 &unlisted,
557 &assessed,
558 provenance,
559 )),
560 None => scan,
561 };
562 let dimensions = QualityDimension::all()
563 .into_iter()
564 .map(|dimension| DimensionSide {
565 dimension,
566 score: at_precision(
567 scores
568 .as_ref()
569 .and_then(|scores| scores.dimension_scores.get(&dimension.to_string()))
570 .copied()
571 .flatten(),
572 ),
573 evidence: evidence(Provenance::Dimension(dimension)),
574 })
575 .collect();
576 Self {
577 reference: ReportReference {
578 id: Some(report.id.clone()),
579 timestamp: Some(report.timestamp.clone()),
580 source: report.source_identifier(),
581 rows_processed: Some(report.execution.rows_processed),
582 evidence: scan,
583 },
584 quality_score: at_precision(scores.as_ref().and_then(|scores| scores.overall_score)),
585 overall_evidence: evidence(Provenance::Overall),
586 dimensions,
587 columns: first_of_each_name(report.column_profiles.iter().map(|column| {
588 let pct = (column.total_count > 0)
589 .then(|| column.null_count as f64 / column.total_count as f64 * 100.0);
590 ColumnSide {
591 name: column.name.clone(),
592 null_pct: at_precision(pct),
593 data_type: Some(column.data_type.clone()),
594 counts: Some((column.total_count, column.null_count)),
595 }
596 })),
597 metric_semantics: report.metric_semantics.clone(),
598 score_weights: quality.map(|assessment| assessment.metrics.score_weights),
603 semantic_hints: (report.schema_version >= 1)
606 .then(|| RecordedHints::from_bindings(&report.semantic_hint_bindings)),
607 identifier_order: (report.schema_version >= 1)
608 .then(|| identifier_order(&report.semantic_hint_bindings)),
609 pattern_locale: report
610 .analysis_settings
611 .as_ref()
612 .map(|settings| settings.pattern_locale),
613 column_selection: report
614 .analysis_settings
615 .as_ref()
616 .map(|settings| settings.columns.clone()),
617 schema_declared: declares_schema(&report.data_source),
618 }
619 }
620
621 pub(crate) fn identifier_order(&self) -> Option<&[String]> {
624 self.identifier_order.as_deref()
625 }
626
627 pub(crate) fn column_selection(&self) -> Option<&ColumnSelection> {
629 self.column_selection.as_ref()
630 }
631
632 pub(crate) fn schema_evidence(&self) -> Evidence {
638 if self.schema_declared {
639 Evidence::Complete
640 } else {
641 self.reference.evidence
642 }
643 }
644
645 pub(crate) fn columns(&self) -> &[ColumnSide] {
647 &self.columns
648 }
649
650 pub fn from_summary_json(text: &str) -> serde_json::Result<Self> {
657 let summary: Summary = serde_json::from_str(text)?;
658 let execution = &summary.execution;
659 let known_gap = scan_coverage(
660 execution.source_exhausted == Some(false) || execution.truncation_reason.is_some(),
661 execution.sampling_applied == Some(true),
662 execution.error_count.is_some_and(|count| count > 0),
663 );
664 let recorded = execution.source_exhausted.is_some()
665 && execution.sampling_applied.is_some()
666 && execution.error_count.is_some();
667 let scan = match known_gap {
670 Evidence::Complete if !recorded => Evidence::Incomplete {
671 reason: EvidenceGap::CoverageUnrecorded,
672 },
673 known => known,
674 };
675 let quality = summary.quality.as_ref();
676 let assessed: Option<Vec<QualityDimension>> = quality.and_then(|quality| {
681 let parsed: Vec<QualityDimension> = quality
682 .assessed_dimensions
683 .iter()
684 .filter_map(|name| name.parse().ok())
689 .collect();
690 (!(parsed.is_empty() && quality.overall_score.is_some())).then_some(parsed)
691 });
692 let evidence = |provenance| {
693 let Some(quality) = quality else {
694 return scan;
695 };
696 let sampled = quality.sampled_dimensions.as_deref();
697 let coverage = match (provenance, &assessed) {
698 (Provenance::Overall, None) if sampled.is_some_and(|s| s.is_empty()) => {
701 Evidence::Complete
702 }
703 (Provenance::Overall, None) => Evidence::Incomplete {
704 reason: EvidenceGap::CoverageUnrecorded,
705 },
706 (_, assessed) => sampled_component_evidence(
710 sampled,
711 &[],
712 assessed.as_deref().unwrap_or_default(),
713 provenance,
714 ),
715 };
716 scan.and(coverage)
717 };
718 let dimensions = QualityDimension::all()
719 .into_iter()
720 .map(|dimension| DimensionSide {
721 dimension,
722 score: at_precision(
723 quality
724 .and_then(|quality| quality.dimension_scores.get(&dimension.to_string()))
725 .copied()
726 .flatten(),
727 ),
728 evidence: evidence(Provenance::Dimension(dimension)),
729 })
730 .collect();
731 Ok(Self {
732 reference: ReportReference {
733 id: None,
734 timestamp: None,
735 source: summary.source,
736 rows_processed: execution.rows_processed,
737 evidence: scan,
738 },
739 quality_score: at_precision(quality.and_then(|quality| quality.overall_score)),
740 overall_evidence: evidence(Provenance::Overall),
741 dimensions,
742 columns: first_of_each_name(summary.columns.into_iter().map(|column| ColumnSide {
743 name: column.name,
744 null_pct: at_precision(column.null_percentage),
745 data_type: column.data_type.as_deref().and_then(DataType::from_name),
748 counts: column.total_count.zip(column.null_count),
749 })),
750 score_weights: quality.and_then(|quality| {
753 quality.score_weights.or_else(|| {
754 summary
755 .metric_semantics
756 .is_some()
757 .then(QualityScoreWeights::default)
758 })
759 }),
760 semantic_hints: summary
761 .schema_version
762 .is_some_and(|version| version >= 1)
763 .then(|| RecordedHints::from_bindings(&summary.semantic_hint_bindings)),
764 identifier_order: summary
765 .schema_version
766 .is_some_and(|version| version >= 1)
767 .then(|| identifier_order(&summary.semantic_hint_bindings)),
768 metric_semantics: summary.metric_semantics,
769 pattern_locale: summary
770 .analysis_settings
771 .as_ref()
772 .map(|settings| settings.pattern_locale),
773 column_selection: summary.analysis_settings.map(|settings| settings.columns),
774 schema_declared: matches!(
777 summary.source_type,
778 Some(SummarySourceType::Dataframe | SummarySourceType::Query)
779 ),
780 })
781 }
782}
783
784fn summary_metric_semantics<'de, D>(deserializer: D) -> Result<Option<MetricSemantics>, D::Error>
787where
788 D: serde::Deserializer<'de>,
789{
790 crate::profile_report::present_value(deserializer).map_err(|error| {
791 <D::Error as serde::de::Error>::custom(format!("metric_semantics: {error}"))
792 })
793}
794
795fn summary_schema_version<'de, D>(deserializer: D) -> Result<Option<u32>, D::Error>
796where
797 D: serde::Deserializer<'de>,
798{
799 crate::profile_report::present_value(deserializer)
800 .map_err(|error| <D::Error as serde::de::Error>::custom(format!("schema_version: {error}")))
801}
802
803fn summary_analysis_settings<'de, D>(deserializer: D) -> Result<Option<AnalysisSettings>, D::Error>
805where
806 D: serde::Deserializer<'de>,
807{
808 crate::profile_report::present_value(deserializer).map_err(|error| {
809 <D::Error as serde::de::Error>::custom(format!("analysis_settings: {error}"))
810 })
811}
812
813fn summary_score_weights<'de, D>(deserializer: D) -> Result<Option<QualityScoreWeights>, D::Error>
815where
816 D: serde::Deserializer<'de>,
817{
818 crate::profile_report::present_value(deserializer).map_err(|error| {
819 <D::Error as serde::de::Error>::custom(format!("quality.score_weights: {error}"))
820 })
821}
822
823#[derive(serde::Deserialize)]
826struct Summary {
827 source: String,
828 #[serde(default)]
830 source_type: Option<SummarySourceType>,
831 execution: SummaryExecution,
832 columns: Vec<SummaryColumn>,
833 quality: Option<SummaryQuality>,
834 #[serde(default, deserialize_with = "summary_metric_semantics")]
835 metric_semantics: Option<MetricSemantics>,
836 #[serde(default, deserialize_with = "summary_schema_version")]
838 schema_version: Option<u32>,
839 #[serde(default)]
840 semantic_hint_bindings: Vec<SemanticHintBinding>,
841 #[serde(default, deserialize_with = "summary_analysis_settings")]
844 analysis_settings: Option<AnalysisSettings>,
845}
846
847#[derive(serde::Deserialize)]
850#[serde(rename_all = "lowercase")]
851enum SummarySourceType {
852 Dataframe,
853 Query,
854 #[serde(other)]
855 Other,
856}
857
858fn declares_schema(source: &DataSource) -> bool {
862 match source {
863 DataSource::File { format, .. } | DataSource::Bytes { format, .. } => {
864 matches!(format, FileFormat::Csv | FileFormat::Parquet)
865 }
866 DataSource::DataFrame { .. } | DataSource::Query { .. } => true,
867 DataSource::Stream { .. } => false,
868 }
869}
870
871#[derive(serde::Deserialize)]
872struct SummaryExecution {
873 rows_processed: Option<usize>,
874 source_exhausted: Option<bool>,
875 truncation_reason: Option<String>,
876 sampling_applied: Option<bool>,
877 error_count: Option<usize>,
878}
879
880#[derive(serde::Deserialize)]
881struct SummaryColumn {
882 name: String,
883 null_percentage: Option<f64>,
884 #[serde(default)]
885 data_type: Option<String>,
886 #[serde(default)]
887 total_count: Option<usize>,
888 #[serde(default)]
889 null_count: Option<usize>,
890}
891
892#[derive(serde::Deserialize)]
893struct SummaryQuality {
894 overall_score: Option<f64>,
895 #[serde(default)]
896 dimension_scores: BTreeMap<String, Option<f64>>,
897 #[serde(default)]
898 assessed_dimensions: Vec<String>,
899 sampled_dimensions: Option<Vec<String>>,
902 #[serde(default, deserialize_with = "summary_score_weights")]
906 score_weights: Option<QualityScoreWeights>,
907}
908
909#[cfg(test)]
910mod tests {
911 use super::*;
912
913 fn complete() -> Sides<Evidence> {
914 Sides {
915 a: Evidence::Complete,
916 b: Evidence::Complete,
917 }
918 }
919
920 #[test]
921 fn a_zero_baseline_has_an_absolute_change_and_no_relative_one() {
922 let change = ScoreChange::new(Some(0.0), Some(50.0), complete());
923 assert_eq!(change.abs, Some(50.0));
924 assert_eq!(change.rel_pct, None);
925 }
926
927 #[test]
928 fn a_change_is_rounded_to_the_serialized_precision() {
929 let change = ScoreChange::new(Some(80.0), Some(80.1), complete());
933 assert_eq!(change.abs, Some(0.1));
934 assert_eq!(change.rel_pct, Some(0.12));
935 }
936
937 #[test]
938 fn a_missing_side_leaves_both_changes_absent() {
939 for (a, b) in [(None, Some(1.0)), (Some(1.0), None), (None, None)] {
940 let change = ScoreChange::new(a, b, complete());
941 assert_eq!((change.abs, change.rel_pct), (None, None));
942 }
943 }
944
945 #[test]
946 fn a_non_finite_input_is_no_number() {
947 assert_eq!(at_precision(Some(f64::NAN)), None);
948 assert_eq!(at_precision(Some(f64::INFINITY)), None);
949 assert_eq!(at_precision(Some(12.345)), Some(12.35));
950 }
951}