dataprof_metrics/analysis/metrics/

mod.rs

//! Data Quality Metrics Calculation Module
//!
//! This module implements comprehensive data quality metric calculations following industry standards.
//! It provides structured assessment across seven key dimensions: Completeness,
//! Consistency, Uniqueness, Accuracy, Timeliness, Validity, and Precision.
//!
//! ## ISO Compliance
//!
//! This module follows:
//! - **ISO 8000-8**: Data Quality (Completeness, Timeliness)
//! - **ISO 8000-61**: Master Data Quality (Consistency)
//! - **ISO 8000-110**: Duplicate Detection (Uniqueness)
//! - **ISO 25012**: Data Quality Model (Accuracy)
//!
//! ## TODO: Future ISO 25012 Dimensions
//!
//! ISO/IEC 25012 defines 15 data quality characteristics. We currently implement 7.
//! Below are candidates for future implementation:
//!
//! ### Inherent Data Quality (content-focused)
//! - **Credibility**: Trustworthiness and believability of data origin
//!   - Impl: Track data source metadata, provenance scoring
//! - **Currentness**: Already partially covered by Timeliness
//!
//! ### System-Dependent Data Quality
//! - **Accessibility**: Data retrievability when needed
//!   - Impl: Response time metrics, availability checks
//! - **Portability**: Transferability across data models/platforms
//!   - Impl: Schema compatibility scoring, format conversion success rate
//! - **Recoverability**: Data integrity during system failures
//!   - Impl: Backup validation, checksum verification
//!
//! ### Research-Proposed Dimensions (MDPI 2024-2025)
//! - **Governance**: Data ownership and responsibility tracking
//! - **Usefulness**: Practical utility scoring for intended purpose
//! - **Semantics**: Meaning clarity and interpretability metrics
//!
//! ### How to Add a New Dimension
//! 1. Create `metrics/{dimension}.rs` with `{Dimension}Metrics` struct + `{Dimension}Calculator`
//! 2. Add `mod {dimension};` and `use {dimension}::...;` in this file
//! 3. Extend `QualityMetrics` in `dataprof-metrics` with new fields
//! 4. Call the calculator in `calculate_comprehensive_metrics()` and map fields
//! 5. Update `IsoQualityConfig` in `dataprof-core` if configurable thresholds are needed
//!
//! ### References
//! - ISO 25012: <https://iso25000.com/index.php/en/iso-25000-standards/iso-25012>
//! - ISO 8000 Wikipedia: <https://en.wikipedia.org/wiki/ISO_8000>
//! - Data Quality 2025 Guide: <https://www.ewsolutions.com/data-quality-quide/>
//! - MDPI Framework Review: <https://www.mdpi.com/2504-2289/9/4/93>

mod accuracy;
mod bounds;
mod completeness;
mod consistency;
mod hint_binding;
mod precision;
#[cfg(test)]
mod testing;
mod timeliness;
mod uniqueness;
pub(crate) mod utils;
mod validity;

// Re-export public types for backward compatibility
pub use bounds::SCORE_BOUNDS_CONFIDENCE;
pub use hint_binding::{compute_value_hint_bindings, value_matches_hint};
pub use utils::{StatisticalValidation, validate_sample_size};

use accuracy::AccuracyCalculator;
use completeness::CompletenessCalculator;
use consistency::ConsistencyCalculator;
use precision::PrecisionCalculator;
use timeliness::TimelinessCalculator;
use uniqueness::UniquenessCalculator;
use validity::ValidityCalculator;

use crate::core::config::IsoQualityConfig;
use crate::core::errors::DataProfilerError;
use crate::types::{
    AccuracyMetrics, ColumnProfile, CompletenessMetrics, ConsistencyMetrics, DataType,
    PrecisionMetrics, QualityDimension, QualityMetrics, RowCompletenessSummary,
    RowDuplicateSummary, TimelinessMetrics, UniquenessMetrics, ValidityMetrics,
};
use dataprof_core::{SemanticHintKind, SemanticHints};
use std::collections::HashMap;

/// Engine for calculating comprehensive data quality metrics.
pub struct MetricsCalculator {
    /// Configurable quality thresholds and score weights.
    pub thresholds: IsoQualityConfig,
    /// Exact full-stream complete-record count, when the engine tracked one.
    row_completeness: Option<RowCompletenessSummary>,
}

impl Default for MetricsCalculator {
    fn default() -> Self {
        Self::new()
    }
}

impl MetricsCalculator {
    /// Create a new calculator with default quality settings.
    pub fn new() -> Self {
        Self::with_thresholds(IsoQualityConfig::default())
    }

    /// Create a calculator with custom thresholds
    pub fn with_thresholds(thresholds: IsoQualityConfig) -> Self {
        Self {
            thresholds,
            row_completeness: None,
        }
    }

    /// Create a calculator with strict thresholds (finance, healthcare)
    pub fn strict() -> Self {
        Self::with_thresholds(IsoQualityConfig::strict())
    }

    /// Create a calculator with lenient thresholds (exploratory, marketing)
    pub fn lenient() -> Self {
        Self::with_thresholds(IsoQualityConfig::lenient())
    }

    /// Supply the engine's exact complete-record count over the full stream.
    ///
    /// Without it the completeness dimension can only report a lower bound
    /// for `complete_records_ratio`, because per-column null totals do not
    /// say whether two nulls shared a record.
    pub fn with_row_completeness(mut self, summary: Option<RowCompletenessSummary>) -> Self {
        self.row_completeness = summary;
        self
    }

    /// Validate statistical requirements for metric calculation
    pub fn validate_sample_size(sample_size: usize, metric_type: &str) -> StatisticalValidation {
        utils::validate_sample_size(sample_size, metric_type)
    }

    /// Check whether a dimension is requested.
    ///
    /// `None` → all dimensions are requested (backward-compatible default).
    fn is_requested(requested: &Option<&[QualityDimension]>, dim: QualityDimension) -> bool {
        match requested {
            None => true,
            Some(dims) => dims.contains(&dim),
        }
    }

    /// Explicit temporal hints, then the columns of declared pairs, plus columns
    /// the profiler itself inferred as dates. Hints come first; inferred columns
    /// are appended in stable profile order and duplicates are removed.
    fn effective_temporal_columns(
        column_profiles: &[ColumnProfile],
        semantic_hints: &SemanticHints,
    ) -> Vec<String> {
        let mut columns: Vec<String> = semantic_hints
            .iter()
            .filter(|(kind, _)| *kind == SemanticHintKind::Temporal)
            .map(|(_, column)| column.to_string())
            .collect();
        for (start, end) in &semantic_hints.temporal_pairs {
            for name in [start, end] {
                if !columns.contains(name) {
                    columns.push(name.clone());
                }
            }
        }
        for profile in column_profiles {
            if profile.data_type == DataType::Date && !columns.contains(&profile.name) {
                columns.push(profile.name.clone());
            }
        }
        columns
    }

    /// Calculate comprehensive data quality metrics from column data.
    ///
    /// When `requested_dimensions` is `None`, all dimensions are computed
    /// (backward-compatible default). When `Some(&[...])`, only the listed
    /// dimensions are computed; the rest are `None` in the returned struct.
    pub fn calculate_comprehensive_metrics(
        &self,
        data: &HashMap<String, Vec<String>>,
        column_profiles: &[ColumnProfile],
        requested_dimensions: Option<&[QualityDimension]>,
    ) -> Result<QualityMetrics, DataProfilerError> {
        self.calculate_comprehensive_metrics_with_hints(
            data,
            column_profiles,
            requested_dimensions,
            &SemanticHints::default(),
            None,
        )
    }

    /// Calculate comprehensive metrics with explicit semantic column hints and,
    /// when the engine tracked whole rows, full-stream duplicate counts that
    /// supersede the sample-based duplicate scan.
    pub fn calculate_comprehensive_metrics_with_hints(
        &self,
        data: &HashMap<String, Vec<String>>,
        column_profiles: &[ColumnProfile],
        requested_dimensions: Option<&[QualityDimension]>,
        semantic_hints: &SemanticHints,
        row_duplicates: Option<RowDuplicateSummary>,
    ) -> Result<QualityMetrics, DataProfilerError> {
        if data.is_empty() {
            return Ok(Self::default_metrics_for_empty_dataset(
                &requested_dimensions,
                self.thresholds.score_weights,
            ));
        }

        let sample_size = Self::calculate_sample_size(data)?;
        let requested = &requested_dimensions;

        // Validate sample size for statistical reliability
        let validation = Self::validate_sample_size(sample_size, "general");
        if !validation.sufficient_sample {
            log::warn!(
                "Sample size ({}) is below recommended minimum ({}) for reliable statistics",
                validation.actual_sample_size,
                validation.min_sample_size
            );
        }

        // Completeness dimension
        let completeness = if Self::is_requested(requested, QualityDimension::Completeness) {
            let c = CompletenessCalculator::new(&self.thresholds)
                .with_row_completeness(self.row_completeness)
                .calculate(data, column_profiles, sample_size)?;
            Some(CompletenessMetrics {
                missing_values_ratio: c.missing_values_ratio,
                complete_records_ratio: c.complete_records_ratio,
                null_columns: c.null_columns,
                total_cells: c.total_cells,
            })
        } else {
            None
        };

        // Consistency dimension
        let consistency = if Self::is_requested(requested, QualityDimension::Consistency) {
            let c = ConsistencyCalculator::calculate(data, column_profiles)?;
            Some(ConsistencyMetrics {
                data_type_consistency: c.data_type_consistency,
                format_violations: c.format_violations,
                encoding_issues: c.encoding_issues,
                values_checked: c.values_checked,
            })
        } else {
            None
        };

        // Uniqueness dimension
        let uniqueness = if Self::is_requested(requested, QualityDimension::Uniqueness) {
            let uniqueness_total_rows = row_duplicates
                .filter(|summary| summary.rows_checked > 0)
                .map_or(sample_size, |summary| summary.rows_checked);
            let u = UniquenessCalculator::new(&self.thresholds).calculate(
                data,
                column_profiles,
                uniqueness_total_rows,
                &semantic_hints.identifier_columns,
                row_duplicates,
            )?;
            Some(UniquenessMetrics {
                duplicate_rows: u.duplicate_rows,
                key_uniqueness: u.key_uniqueness,
                high_cardinality_warning: u.high_cardinality_warning,
                rows_checked: u.rows_checked,
                key_column: u.key_column,
                key_values_checked: u.key_values_checked,
                duplicate_rows_approximate: u.duplicate_rows_approximate,
            })
        } else {
            None
        };

        // Accuracy dimension
        let accuracy = if Self::is_requested(requested, QualityDimension::Accuracy) {
            let accuracy_calculator = AccuracyCalculator::new(&self.thresholds);
            let a = if semantic_hints.positive_columns.is_empty() {
                accuracy_calculator.calculate(data, column_profiles)?
            } else {
                accuracy_calculator.calculate_with_positive_columns(
                    data,
                    column_profiles,
                    &semantic_hints.positive_columns,
                )?
            };
            Some(AccuracyMetrics {
                outlier_ratio: a.outlier_ratio,
                outlier_values_checked: Some(a.outlier_values_checked),
                range_violations: a.range_violations,
                negative_values_in_positive: a.negative_values_in_positive,
                numeric_values_checked: a.numeric_values_checked,
                rule_values_checked: a.rule_values_checked,
                rule_violating_values: a.rule_violating_values,
            })
        } else {
            None
        };

        // Timeliness dimension
        let timeliness = if Self::is_requested(requested, QualityDimension::Timeliness) {
            let temporal_columns =
                Self::effective_temporal_columns(column_profiles, semantic_hints);
            let t = TimelinessCalculator::new(&self.thresholds).calculate(
                data,
                &temporal_columns,
                &semantic_hints.temporal_pairs,
                column_profiles,
            )?;
            Some(TimelinessMetrics {
                future_dates_count: t.future_dates_count,
                stale_data_ratio: t.stale_data_ratio,
                temporal_violations: t.temporal_violations,
                invalid_date_values: t.invalid_date_values,
                date_values_checked: t.date_values_checked,
                temporal_pairs_checked: t.temporal_pairs_checked,
            })
        } else {
            None
        };

        let validity = if Self::is_requested(requested, QualityDimension::Validity) {
            let validity = ValidityCalculator::calculate(data, column_profiles);
            Some(ValidityMetrics {
                valid_values_ratio: validity.valid_values_ratio,
                invalid_values: validity.invalid_values,
                values_checked: validity.values_checked,
            })
        } else {
            None
        };

        let precision = if Self::is_requested(requested, QualityDimension::Precision) {
            let precision = PrecisionCalculator::calculate(data, column_profiles);
            Some(PrecisionMetrics {
                decimal_places_consistency: precision.decimal_places_consistency,
                inconsistent_precision_values: precision.inconsistent_precision_values,
                numeric_values_checked: precision.numeric_values_checked,
            })
        } else {
            None
        };

        Ok(QualityMetrics {
            completeness,
            consistency,
            uniqueness,
            accuracy,
            timeliness,
            validity,
            precision,
            low_sample_warning: !validation.sufficient_sample,
            score_weights: self.thresholds.score_weights,
        })
    }

    /// Create default metrics for empty dataset (only requested dimensions are populated).
    fn default_metrics_for_empty_dataset(
        requested: &Option<&[QualityDimension]>,
        score_weights: dataprof_core::QualityScoreWeights,
    ) -> QualityMetrics {
        let is_req = |d| match requested {
            None => true,
            Some(dims) => dims.contains(&d),
        };

        QualityMetrics {
            completeness: if is_req(QualityDimension::Completeness) {
                Some(CompletenessMetrics {
                    missing_values_ratio: 0.0,
                    complete_records_ratio: 100.0,
                    null_columns: vec![],
                    total_cells: 0,
                })
            } else {
                None
            },
            consistency: if is_req(QualityDimension::Consistency) {
                Some(ConsistencyMetrics {
                    data_type_consistency: 100.0,
                    format_violations: 0,
                    encoding_issues: 0,
                    values_checked: 0,
                })
            } else {
                None
            },
            uniqueness: if is_req(QualityDimension::Uniqueness) {
                Some(UniquenessMetrics {
                    duplicate_rows: 0,
                    key_uniqueness: 100.0,
                    high_cardinality_warning: false,
                    rows_checked: 0,
                    key_column: None,
                    key_values_checked: None,
                    duplicate_rows_approximate: false,
                })
            } else {
                None
            },
            accuracy: if is_req(QualityDimension::Accuracy) {
                Some(AccuracyMetrics {
                    outlier_ratio: 0.0,
                    outlier_values_checked: Some(0),
                    range_violations: 0,
                    negative_values_in_positive: 0,
                    numeric_values_checked: 0,
                    rule_values_checked: 0,
                    rule_violating_values: 0,
                })
            } else {
                None
            },
            timeliness: if is_req(QualityDimension::Timeliness) {
                Some(TimelinessMetrics {
                    future_dates_count: 0,
                    stale_data_ratio: 0.0,
                    temporal_violations: 0,
                    invalid_date_values: 0,
                    date_values_checked: 0,
                    temporal_pairs_checked: 0,
                })
            } else {
                None
            },
            validity: if is_req(QualityDimension::Validity) {
                Some(ValidityMetrics {
                    valid_values_ratio: 100.0,
                    invalid_values: 0,
                    values_checked: 0,
                })
            } else {
                None
            },
            precision: if is_req(QualityDimension::Precision) {
                Some(PrecisionMetrics {
                    decimal_places_consistency: 100.0,
                    inconsistent_precision_values: 0,
                    numeric_values_checked: 0,
                })
            } else {
                None
            },
            low_sample_warning: false,
            score_weights,
        }
    }

    /// Calculate quality metrics with bifurcated computation for streaming.
    ///
    /// **Phase A (exact from global counters)**: Completeness metrics are computed
    /// from `ColumnProfile` stats (`null_count`, `total_count`) which are exact
    /// even for infinite streams. Key uniqueness already uses `ColumnProfile`.
    ///
    /// **Phase B (sampled)**: Consistency, Accuracy, and Timeliness are computed
    /// from the bounded reservoir sample. Duplicate rows use a full-stream row
    /// tracker when supplied; otherwise they fall back to an aligned sample.
    ///
    /// Returns a [`BifurcatedResult`] containing the metrics plus provenance
    /// for which dimensions are exact vs sampled.
    pub fn calculate_bifurcated_metrics(
        &self,
        data: &HashMap<String, Vec<String>>,
        column_profiles: &[ColumnProfile],
        requested_dimensions: Option<&[QualityDimension]>,
    ) -> Result<BifurcatedResult, DataProfilerError> {
        self.calculate_bifurcated_metrics_with_hints(
            data,
            column_profiles,
            requested_dimensions,
            &SemanticHints::default(),
            None,
        )
    }

    /// Calculate bifurcated metrics with explicit semantic column hints and,
    /// when the engine tracked whole rows, full-stream duplicate counts that
    /// supersede the sample-based duplicate scan.
    pub fn calculate_bifurcated_metrics_with_hints(
        &self,
        data: &HashMap<String, Vec<String>>,
        column_profiles: &[ColumnProfile],
        requested_dimensions: Option<&[QualityDimension]>,
        semantic_hints: &SemanticHints,
        row_duplicates: Option<RowDuplicateSummary>,
    ) -> Result<BifurcatedResult, DataProfilerError> {
        if data.is_empty() && column_profiles.is_empty() {
            return Ok(BifurcatedResult {
                metrics: Self::default_metrics_for_empty_dataset(
                    &requested_dimensions,
                    self.thresholds.score_weights,
                ),
                exact_dimensions: vec![],
                sampled_dimensions: vec![],
                sample_size: 0,
                score_bounds: None,
            });
        }

        let total_rows = column_profiles.first().map(|p| p.total_count).unwrap_or(0);
        let sample_rows = Self::calculate_sample_size(data).unwrap_or(0);
        let requested = &requested_dimensions;
        // One calculator, so the score bounds judge dates against the same
        // reference day as the score.
        let timeliness_calculator = TimelinessCalculator::new(&self.thresholds);
        let temporal_columns = Self::effective_temporal_columns(column_profiles, semantic_hints);

        let mut exact_dimensions = Vec::new();
        let mut sampled_dimensions = Vec::new();

        // Phase A: Completeness from exact global counters
        let completeness = if Self::is_requested(requested, QualityDimension::Completeness) {
            let c = CompletenessCalculator::new(&self.thresholds)
                .with_row_completeness(self.row_completeness)
                .calculate_from_profiles(column_profiles)?;
            exact_dimensions.push("completeness".to_string());
            Some(CompletenessMetrics {
                missing_values_ratio: c.missing_values_ratio,
                complete_records_ratio: c.complete_records_ratio,
                null_columns: c.null_columns,
                total_cells: c.total_cells,
            })
        } else {
            None
        };

        // Phase B: Sampled dimensions from reservoir data
        let consistency = if Self::is_requested(requested, QualityDimension::Consistency) {
            let c = if !data.is_empty() {
                ConsistencyCalculator::calculate(data, column_profiles)?
            } else {
                consistency::ConsistencyMetrics {
                    data_type_consistency: 100.0,
                    format_violations: 0,
                    encoding_issues: 0,
                    values_checked: 0,
                    inconsistent_values: 0,
                }
            };
            sampled_dimensions.push("consistency".to_string());
            Some(ConsistencyMetrics {
                data_type_consistency: c.data_type_consistency,
                format_violations: c.format_violations,
                encoding_issues: c.encoding_issues,
                values_checked: c.values_checked,
            })
        } else {
            None
        };

        let uniqueness = if Self::is_requested(requested, QualityDimension::Uniqueness) {
            let u = UniquenessCalculator::new(&self.thresholds).calculate(
                data,
                column_profiles,
                total_rows,
                &semantic_hints.identifier_columns,
                row_duplicates,
            )?;
            // Only claim provenance for components that carry signal:
            // key_uniqueness means nothing without an identified key column,
            // and the duplicate scan refuses misaligned per-column samples
            // (rows_checked == 0). Full-stream tracker counts are exact
            // until the distinct estimator spills to its HLL sketch, after
            // which they are flagged approximate and filed as non-exact.
            // Key uniqueness is the key column's distinct count over its
            // non-null values, so it is only as exact as that count: an
            // estimated count is filed as sampled, and the score says it
            // rests on an estimate rather than presenting one as a
            // measurement. A key with no value measured nothing.
            let metrics = UniquenessMetrics {
                duplicate_rows: u.duplicate_rows,
                key_uniqueness: u.key_uniqueness,
                high_cardinality_warning: u.high_cardinality_warning,
                rows_checked: u.rows_checked,
                key_column: u.key_column,
                key_values_checked: u.key_values_checked,
                duplicate_rows_approximate: u.duplicate_rows_approximate,
            };
            if let Some(key) = metrics
                .key_column
                .as_deref()
                .filter(|_| metrics.key_measured())
            {
                let estimated = column_profiles
                    .iter()
                    .find(|profile| profile.name == key)
                    .is_some_and(|profile| profile.unique_count_is_approximate == Some(true));
                if estimated {
                    sampled_dimensions.push("key_uniqueness".to_string());
                } else {
                    exact_dimensions.push("key_uniqueness".to_string());
                }
            }
            if u.rows_checked > 0 {
                let from_tracker = row_duplicates.is_some_and(|s| s.rows_checked > 0);
                if from_tracker && !u.duplicate_rows_approximate {
                    exact_dimensions.push("duplicate_rows".to_string());
                } else {
                    sampled_dimensions.push("duplicate_rows".to_string());
                }
            }
            Some(metrics)
        } else {
            None
        };

        let accuracy = if Self::is_requested(requested, QualityDimension::Accuracy) {
            let a = if !data.is_empty() {
                AccuracyCalculator::new(&self.thresholds).calculate_with_positive_columns(
                    data,
                    column_profiles,
                    &semantic_hints.positive_columns,
                )?
            } else {
                accuracy::AccuracyMetrics {
                    outlier_ratio: 0.0,
                    range_violations: 0,
                    negative_values_in_positive: 0,
                    numeric_values_checked: 0,
                    rule_values_checked: 0,
                    rule_violating_values: 0,
                    outliers: 0,
                    outlier_values_checked: 0,
                }
            };
            sampled_dimensions.push("accuracy".to_string());
            Some(AccuracyMetrics {
                outlier_ratio: a.outlier_ratio,
                outlier_values_checked: Some(a.outlier_values_checked),
                range_violations: a.range_violations,
                negative_values_in_positive: a.negative_values_in_positive,
                numeric_values_checked: a.numeric_values_checked,
                rule_values_checked: a.rule_values_checked,
                rule_violating_values: a.rule_violating_values,
            })
        } else {
            None
        };

        let timeliness = if Self::is_requested(requested, QualityDimension::Timeliness) {
            let t = if !data.is_empty() {
                timeliness_calculator.calculate(
                    data,
                    &temporal_columns,
                    &semantic_hints.temporal_pairs,
                    column_profiles,
                )?
            } else {
                timeliness::TimelinessMetrics {
                    future_dates_count: 0,
                    stale_data_ratio: 0.0,
                    temporal_violations: 0,
                    invalid_date_values: 0,
                    date_values_checked: 0,
                    temporal_pairs_checked: 0,
                    valid_dates: 0,
                    stale_dates: 0,
                    temporal_column_pairs: 0,
                    compared_pairs: Vec::new(),
                }
            };
            sampled_dimensions.push("timeliness".to_string());
            Some(TimelinessMetrics {
                future_dates_count: t.future_dates_count,
                stale_data_ratio: t.stale_data_ratio,
                temporal_violations: t.temporal_violations,
                invalid_date_values: t.invalid_date_values,
                date_values_checked: t.date_values_checked,
                temporal_pairs_checked: t.temporal_pairs_checked,
            })
        } else {
            None
        };

        let validity = if Self::is_requested(requested, QualityDimension::Validity) {
            let validity = ValidityCalculator::calculate(data, column_profiles);
            sampled_dimensions.push("validity".to_string());
            Some(ValidityMetrics {
                valid_values_ratio: validity.valid_values_ratio,
                invalid_values: validity.invalid_values,
                values_checked: validity.values_checked,
            })
        } else {
            None
        };

        let precision = if Self::is_requested(requested, QualityDimension::Precision) {
            let precision = PrecisionCalculator::calculate(data, column_profiles);
            sampled_dimensions.push("precision".to_string());
            Some(PrecisionMetrics {
                decimal_places_consistency: precision.decimal_places_consistency,
                inconsistent_precision_values: precision.inconsistent_precision_values,
                numeric_values_checked: precision.numeric_values_checked,
            })
        } else {
            None
        };

        // total_rows reflects the full stream — that's what the user cares about
        // for whether the quality metrics are reliable. Fall back to sample_rows
        // when total_rows is unknown (e.g. profile arrays mode).
        let reliability_sample = if total_rows > 0 {
            total_rows
        } else {
            sample_rows
        };
        let validation = Self::validate_sample_size(reliability_sample, "general");

        let metrics = QualityMetrics {
            completeness,
            consistency,
            uniqueness,
            accuracy,
            timeliness,
            validity,
            precision,
            low_sample_warning: !validation.sufficient_sample,
            score_weights: self.thresholds.score_weights,
        };
        let score_bounds = bounds::score_bounds(
            &self.thresholds,
            &timeliness_calculator,
            data,
            column_profiles,
            semantic_hints,
            &temporal_columns,
            &metrics,
            &sampled_dimensions,
            row_duplicates,
        )?;

        Ok(BifurcatedResult {
            metrics,
            exact_dimensions,
            sampled_dimensions,
            sample_size: sample_rows,
            score_bounds,
        })
    }

    /// Calculate sample size across data column arrays (using max inner length heuristic)
    fn calculate_sample_size(
        data: &HashMap<String, Vec<String>>,
    ) -> Result<usize, DataProfilerError> {
        data.values().map(|v| v.len()).max().ok_or_else(|| {
            DataProfilerError::MetricsCalculationError {
                message: "No data columns found".to_string(),
            }
        })
    }
}

/// Result of bifurcated quality metric calculation.
///
/// Contains the computed metrics plus provenance information about which
/// dimensions were computed exactly (from global streaming counters) and
/// which were computed from a bounded sample.
pub struct BifurcatedResult {
    /// The computed quality metrics
    pub metrics: QualityMetrics,
    /// Dimensions computed from exact global counters (e.g., "completeness", "key_uniqueness")
    pub exact_dimensions: Vec<String>,
    /// Dimensions computed from the bounded reservoir sample (e.g., "consistency", "accuracy")
    pub sampled_dimensions: Vec<String>,
    /// Number of sample rows used for Phase B dimensions
    pub sample_size: usize,
    /// Where the scores over every scanned row lie, when some dimension was sampled.
    pub score_bounds: Option<crate::quality::ScoreBounds>,
}

#[cfg(test)]
mod tests {
    use super::testing::string_profile;
    use super::*;
    use crate::types::{ColumnStats, DataType};
    use dataprof_core::QualityScoreWeights;

    #[test]
    fn comprehensive_metrics_use_tracker_rows_for_cardinality_ratio() {
        let data = HashMap::from([(
            "city".to_string(),
            vec!["Rome".to_string(), "Milan".to_string()],
        )]);
        let profiles = vec![ColumnProfile {
            name: "city".to_string(),
            data_type: DataType::String,
            null_count: 0,
            total_count: 1_000,
            unique_count: Some(100),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            invalid_count: None,
            type_homogeneity: None,
            locale_number_count: None,
            stats: ColumnStats::None,
            patterns: Some(vec![]),
        }];
        let requested = [QualityDimension::Uniqueness];

        let metrics = MetricsCalculator::new()
            .calculate_comprehensive_metrics_with_hints(
                &data,
                &profiles,
                Some(&requested),
                &SemanticHints::default(),
                Some(RowDuplicateSummary {
                    duplicate_rows: 0,
                    rows_checked: 1_000,
                    approximate: false,
                    max_duplicate_rows: 0,
                }),
            )
            .expect("quality metrics");

        assert!(
            !metrics
                .uniqueness
                .expect("uniqueness metrics")
                .high_cardinality_warning,
            "100 distinct values out of 1,000 rows is not high cardinality"
        );
    }

    #[test]
    fn an_estimated_key_count_is_filed_as_sampled() {
        let key = |approximate: bool| ColumnProfile {
            name: "order_id".to_string(),
            data_type: DataType::String,
            null_count: 0,
            total_count: 5_000_000,
            unique_count: Some(4_990_000),
            unique_count_is_approximate: Some(approximate),
            unique_count_lower_bound: None,
            invalid_count: None,
            type_homogeneity: None,
            locale_number_count: None,
            stats: ColumnStats::None,
            patterns: Some(vec![]),
        };
        let data = HashMap::from([("order_id".to_string(), vec!["A-1".to_string()])]);
        let requested = [QualityDimension::Uniqueness];

        for (approximate, filed_as_sampled) in [(true, true), (false, false)] {
            let result = MetricsCalculator::new()
                .calculate_bifurcated_metrics_with_hints(
                    &data,
                    &[key(approximate)],
                    Some(&requested),
                    &SemanticHints::new(Vec::new(), vec!["order_id".to_string()]),
                    None,
                )
                .expect("quality metrics");
            assert_eq!(
                result
                    .sampled_dimensions
                    .contains(&"key_uniqueness".to_string()),
                filed_as_sampled,
                "approximate key count: {approximate}"
            );
            assert_eq!(
                result
                    .exact_dimensions
                    .contains(&"key_uniqueness".to_string()),
                !filed_as_sampled
            );
        }
    }

    #[test]
    fn a_key_holding_no_value_claims_no_provenance() {
        // The key was identified but every value was null: it measured
        // nothing, so neither list may say how it was measured (#1005).
        let key = ColumnProfile {
            null_count: 5_000,
            unique_count: Some(0),
            unique_count_is_approximate: Some(false),
            ..string_profile("order_id", 5_000, 5_000)
        };
        let result = MetricsCalculator::new()
            .calculate_bifurcated_metrics_with_hints(
                &HashMap::from([("order_id".to_string(), vec![])]),
                &[key],
                Some(&[QualityDimension::Uniqueness]),
                &SemanticHints::new(Vec::new(), vec!["order_id".to_string()]),
                None,
            )
            .expect("quality metrics");
        let uniqueness = result.metrics.uniqueness.expect("uniqueness metrics");
        assert_eq!(uniqueness.key_column.as_deref(), Some("order_id"));
        assert_eq!(uniqueness.key_values_checked, Some(0));
        let label = "key_uniqueness".to_string();
        assert!(!result.exact_dimensions.contains(&label));
        assert!(!result.sampled_dimensions.contains(&label));
    }

    #[test]
    fn calculator_copies_custom_score_weights_into_metrics() {
        let weights = QualityScoreWeights {
            completeness: 1.0,
            consistency: 0.0,
            uniqueness: 0.0,
            accuracy: 0.0,
            timeliness: 0.0,
            validity: 0.0,
            precision: 0.0,
        };
        let config = IsoQualityConfig {
            score_weights: weights,
            ..IsoQualityConfig::default()
        };
        let data = HashMap::from([("value".to_string(), vec!["x".to_string()])]);
        let profiles = vec![ColumnProfile {
            name: "value".to_string(),
            data_type: DataType::String,
            null_count: 0,
            total_count: 1,
            unique_count: Some(1),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            invalid_count: None,
            type_homogeneity: None,
            locale_number_count: None,
            stats: ColumnStats::None,
            patterns: Some(vec![]),
        }];

        let metrics = MetricsCalculator::with_thresholds(config)
            .calculate_comprehensive_metrics(&data, &profiles, None)
            .expect("quality metrics");

        assert_eq!(metrics.score_weights, weights);
    }

    #[test]
    fn unrequested_completeness_remains_absent() {
        let data = HashMap::from([("value".to_string(), vec!["x".to_string()])]);
        let profiles = vec![string_profile("value", 1, 0)];
        let requested = [QualityDimension::Consistency];

        let metrics = MetricsCalculator::new()
            .calculate_comprehensive_metrics(&data, &profiles, Some(&requested))
            .expect("quality metrics");

        assert!(
            metrics.completeness.is_none(),
            "an unrequested dimension must remain absent rather than looking computed"
        );
        assert!(
            metrics.consistency.is_some(),
            "the requested dimension should still be computed"
        );
    }

    #[test]
    fn empty_bifurcated_sample_keeps_new_dimensions_neutral_and_unassessed() {
        let profiles = vec![ColumnProfile {
            name: "amount".to_string(),
            data_type: DataType::Float,
            null_count: 0,
            total_count: 100,
            unique_count: Some(10),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            invalid_count: None,
            type_homogeneity: None,
            locale_number_count: None,
            stats: ColumnStats::None,
            patterns: Some(vec![]),
        }];
        let requested = [QualityDimension::Validity, QualityDimension::Precision];

        let result = MetricsCalculator::new()
            .calculate_bifurcated_metrics(&HashMap::new(), &profiles, Some(&requested))
            .expect("bifurcated quality metrics");
        let validity = result.metrics.validity.as_ref().expect("validity metrics");
        let precision = result
            .metrics
            .precision
            .as_ref()
            .expect("precision metrics");

        assert_eq!(validity.valid_values_ratio, 100.0);
        assert_eq!(validity.invalid_values, 0);
        assert_eq!(validity.values_checked, 0);
        assert_eq!(precision.decimal_places_consistency, 100.0);
        assert_eq!(precision.inconsistent_precision_values, 0);
        assert_eq!(precision.numeric_values_checked, 0);
        assert!(result.metrics.assessed_dimensions().is_empty());
    }
}