dataprof-metrics 0.10.0

//! 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 completeness;
mod consistency;
mod hint_binding;
mod precision;
#[cfg(test)]
mod testing;
mod timeliness;
mod uniqueness;
mod utils;
mod validity;

// Re-export public types for backward compatibility
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, RowDuplicateSummary, TimelinessMetrics,
    UniquenessMetrics, ValidityMetrics,
};
use dataprof_core::SemanticHints;
use std::collections::HashMap;

/// Engine for calculating comprehensive data quality metrics.
pub struct MetricsCalculator {
    /// Configurable quality thresholds and score weights.
    pub thresholds: IsoQualityConfig,
}

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

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

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

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

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

    /// 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 plus columns the profiler itself inferred as dates.
    /// Hint order wins for ambiguous start/end role matching; 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 = semantic_hints.temporal_columns.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_positive_columns(
            data,
            column_profiles,
            requested_dimensions,
            &[],
            None,
        )
    }

    /// Calculate comprehensive metrics with explicit positive-only 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_positive_columns(
        &self,
        data: &HashMap<String, Vec<String>>,
        column_profiles: &[ColumnProfile],
        requested_dimensions: Option<&[QualityDimension]>,
        positive_columns: &[String],
        row_duplicates: Option<RowDuplicateSummary>,
    ) -> Result<QualityMetrics, DataProfilerError> {
        self.calculate_comprehensive_metrics_with_semantic_hints(
            data,
            column_profiles,
            requested_dimensions,
            positive_columns,
            &[],
            row_duplicates,
        )
    }

    /// Calculate comprehensive metrics with explicit semantic column hints.
    pub fn calculate_comprehensive_metrics_with_semantic_hints(
        &self,
        data: &HashMap<String, Vec<String>>,
        column_profiles: &[ColumnProfile],
        requested_dimensions: Option<&[QualityDimension]>,
        positive_columns: &[String],
        identifier_columns: &[String],
        row_duplicates: Option<RowDuplicateSummary>,
    ) -> Result<QualityMetrics, DataProfilerError> {
        let semantic_hints =
            SemanticHints::new(positive_columns.to_vec(), identifier_columns.to_vec());
        self.calculate_comprehensive_metrics_with_all_semantic_hints(
            data,
            column_profiles,
            requested_dimensions,
            &semantic_hints,
            row_duplicates,
        )
    }

    /// Calculate comprehensive metrics with all explicit semantic column hints.
    pub fn calculate_comprehensive_metrics_with_all_semantic_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).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,
                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,
                range_violations: a.range_violations,
                negative_values_in_positive: a.negative_values_in_positive,
                numeric_values_checked: a.numeric_values_checked,
            })
        } 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)?;
            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,
                    duplicate_rows_approximate: false,
                })
            } else {
                None
            },
            accuracy: if is_req(QualityDimension::Accuracy) {
                Some(AccuracyMetrics {
                    outlier_ratio: 0.0,
                    range_violations: 0,
                    negative_values_in_positive: 0,
                    numeric_values_checked: 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_positive_columns(
            data,
            column_profiles,
            requested_dimensions,
            &[],
            None,
        )
    }

    /// Calculate bifurcated metrics with explicit positive-only 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_positive_columns(
        &self,
        data: &HashMap<String, Vec<String>>,
        column_profiles: &[ColumnProfile],
        requested_dimensions: Option<&[QualityDimension]>,
        positive_columns: &[String],
        row_duplicates: Option<RowDuplicateSummary>,
    ) -> Result<BifurcatedResult, DataProfilerError> {
        self.calculate_bifurcated_metrics_with_semantic_hints(
            data,
            column_profiles,
            requested_dimensions,
            positive_columns,
            &[],
            row_duplicates,
        )
    }

    /// Calculate bifurcated metrics with explicit semantic column hints.
    pub fn calculate_bifurcated_metrics_with_semantic_hints(
        &self,
        data: &HashMap<String, Vec<String>>,
        column_profiles: &[ColumnProfile],
        requested_dimensions: Option<&[QualityDimension]>,
        positive_columns: &[String],
        identifier_columns: &[String],
        row_duplicates: Option<RowDuplicateSummary>,
    ) -> Result<BifurcatedResult, DataProfilerError> {
        let semantic_hints =
            SemanticHints::new(positive_columns.to_vec(), identifier_columns.to_vec());
        self.calculate_bifurcated_metrics_with_all_semantic_hints(
            data,
            column_profiles,
            requested_dimensions,
            &semantic_hints,
            row_duplicates,
        )
    }

    /// Calculate bifurcated metrics with all explicit semantic column hints.
    pub fn calculate_bifurcated_metrics_with_all_semantic_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,
            });
        }

        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;

        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)
                .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,
                }
            };
            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.
            if u.key_column.is_some() {
                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(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,
                duplicate_rows_approximate: u.duplicate_rows_approximate,
            })
        } 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,
                }
            };
            sampled_dimensions.push("accuracy".to_string());
            Some(AccuracyMetrics {
                outlier_ratio: a.outlier_ratio,
                range_violations: a.range_violations,
                negative_values_in_positive: a.negative_values_in_positive,
                numeric_values_checked: a.numeric_values_checked,
            })
        } else {
            None
        };

        let timeliness = if Self::is_requested(requested, QualityDimension::Timeliness) {
            let temporal_columns =
                Self::effective_temporal_columns(column_profiles, semantic_hints);
            let t = if !data.is_empty() {
                TimelinessCalculator::new(&self.thresholds).calculate(data, &temporal_columns)?
            } 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,
                }
            };
            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,
        };

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

    /// 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,
}

#[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),
            invalid_count: None,
            stats: ColumnStats::None,
            patterns: Some(vec![]),
        }];
        let requested = [QualityDimension::Uniqueness];

        let metrics = MetricsCalculator::new()
            .calculate_comprehensive_metrics_with_positive_columns(
                &data,
                &profiles,
                Some(&requested),
                &[],
                Some(RowDuplicateSummary {
                    duplicate_rows: 0,
                    rows_checked: 1_000,
                    approximate: false,
                }),
            )
            .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 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),
            invalid_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),
            invalid_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());
    }
}