dataprof-runtime 0.13.1

//! Shared conversion from [`StreamingColumnCollection`] / [`StreamingStatistics`]
//! into [`ColumnProfile`] and quality-check sample maps.
//!
//! All engines that need to produce a [`ColumnProfile`] should call
//! [`build_column_profile`] instead of constructing one manually.
//! This ensures consistent stats calculation and pattern detection.

use std::collections::HashMap;

use dataprof_core::{
    BooleanStats, ColumnProfile, ColumnStats, DataType, DateTimeStats, Locale, SemanticHints,
    TextStats,
};
use dataprof_metrics::{
    analysis::column::stats_over_values,
    analysis::inference::{
        classify_lexical_forms, count_locale_numbers, holds_digit_codes, is_inferred_date_token,
        is_integer_token, is_null_like_token, parse_strict_boolean_token,
    },
    calculate_datetime_stats, calculate_text_stats, detect_patterns,
    stats::numeric::{calculate_coefficient_of_variation, compute_numeric_stats_with_parsed_count},
};

use crate::streaming_stats::{StreamingColumnCollection, StreamingStatistics};

/// Inputs that every engine can provide for centralized profile construction.
pub struct ColumnProfileInput<'a> {
    pub name: String,
    pub data_type: DataType,
    pub total_count: usize,
    pub null_count: usize,
    pub unique_count: Option<usize>,
    /// Whether `unique_count` is an approximate (HLL) estimate. `None` mirrors
    /// `unique_count: None`; `Some(false)` for exact counts; `Some(true)` once
    /// the engine's cardinality estimator has spilled to its HLL sketch.
    pub unique_count_is_approximate: Option<bool>,
    /// See [`ColumnProfile::unique_count_lower_bound`]: the estimator's
    /// [`distinct_lower_bound`](dataprof_metrics::CardinalityEstimator::distinct_lower_bound),
    /// or `None` when the caller has no estimator behind the count.
    pub unique_count_lower_bound: Option<usize>,
    pub sample_values: &'a [String],
    /// Pre-computed text lengths for engines that track them incrementally.
    /// When `Some`, text stats are built from these instead of re-scanning samples.
    pub text_lengths: Option<TextLengths>,
    /// Pre-computed boolean counts (true_count, false_count) for boolean columns.
    pub boolean_counts: Option<(usize, usize)>,
    /// When true, skip statistics computation (produce `ColumnStats::None`).
    pub skip_statistics: bool,
    /// When true, skip pattern detection (produce `patterns: None`).
    pub skip_patterns: bool,
    /// Optional locale for pattern detection (e.g. "IT", "US").
    pub locale: Option<Locale>,
    /// Exact aggregates over every numeric value the engine streamed.
    ///
    /// When present, these override the sample-derived `min`, `max`, `mean`,
    /// `std_dev`, and `variance` on numeric columns, so those fields stay
    /// exact even when `sample_values` no longer covers the full stream.
    /// `None` means `sample_values` *is* the full data (in-memory sources)
    /// or the engine has no exact accumulators for the column.
    pub exact_numeric: Option<ExactNumericAggregates>,
    /// Number of values over the full stream accepted by the temporal parser.
    /// `None` means `sample_values` is the complete in-memory column.
    pub exact_date_matches: Option<usize>,
}

/// Exact streaming aggregates for a numeric column: O(1)-memory statistics that an
/// engine computed over the entire stream, as opposed to the bounded
/// `sample_values` it retained.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct ExactNumericAggregates {
    pub min: f64,
    pub max: f64,
    pub mean: f64,
    /// Standard deviation with the unbiased (n-1) denominator.
    pub std_dev: f64,
    /// Variance with the unbiased (n-1) denominator.
    pub variance: f64,
    /// Number of parsed numeric values covered by these aggregates.
    pub count: usize,
}

/// Pre-computed text length stats from streaming/columnar engines.
pub struct TextLengths {
    pub min_length: usize,
    pub max_length: usize,
    pub avg_length: f64,
}

/// Build a [`ColumnProfile`] from engine-agnostic inputs.
///
/// This is the single canonical construction path. Engines provide
/// pre-inferred `DataType`, counters, sample values, and optionally
/// pre-computed text lengths; this function handles stats calculation
/// and pattern detection.
pub fn build_column_profile(input: ColumnProfileInput<'_>) -> ColumnProfile {
    if input.data_type == DataType::Nested {
        return nested_column_profile(input.name, input.total_count, input.null_count);
    }

    let mut invalid_count = None;
    let stats = if input.skip_statistics {
        ColumnStats::None
    } else {
        match input.data_type {
            DataType::Integer | DataType::Float => {
                let (mut numeric, sampled_numeric) =
                    compute_numeric_stats_with_parsed_count(input.sample_values);
                // Values the statistics actually cover. With exact stream
                // aggregates that is the engine's full parsed count; without
                // them the sample *is* the full data (in-memory sources).
                let parsed_total = match &input.exact_numeric {
                    Some(exact) => exact.count,
                    None => sampled_numeric,
                };
                invalid_count = Some(
                    input
                        .total_count
                        .saturating_sub(input.null_count)
                        .saturating_sub(parsed_total),
                );
                if let Some(exact) = &input.exact_numeric {
                    numeric.min = exact.min;
                    numeric.max = exact.max;
                    numeric.mean = exact.mean;
                    numeric.std_dev = exact.std_dev.is_finite().then_some(exact.std_dev);
                    numeric.variance = exact.variance.is_finite().then_some(exact.variance);
                    numeric.coefficient_of_variation = numeric
                        .std_dev
                        .and_then(|value| calculate_coefficient_of_variation(value, exact.mean));
                    if numeric.std_dev.is_none() {
                        numeric.skewness = None;
                        numeric.kurtosis = None;
                    }
                    // Order statistics (median, quartiles, mode, skewness,
                    // kurtosis, outliers) still come from the retained sample;
                    // disclose that whenever the sample no longer covers every
                    // numeric value the exact aggregates saw.
                    if exact.count > sampled_numeric {
                        numeric.is_approximate = Some(true);
                    }
                }
                if parsed_total == 0 {
                    ColumnStats::None
                } else {
                    ColumnStats::Numeric(numeric)
                }
            }
            DataType::Date => {
                let parsed_dates = input.exact_date_matches.unwrap_or_else(|| {
                    input
                        .sample_values
                        .iter()
                        .filter(|value| {
                            dataprof_metrics::value_matches_hint(
                                value,
                                dataprof_core::SemanticHintKind::Temporal,
                            )
                        })
                        .count()
                });
                invalid_count = Some(
                    input
                        .total_count
                        .saturating_sub(input.null_count)
                        .saturating_sub(parsed_dates),
                );
                if !input.sample_values.is_empty() {
                    calculate_datetime_stats(input.sample_values)
                } else if let Some(tl) = &input.text_lengths {
                    ColumnStats::Text(TextStats::from_lengths(
                        tl.min_length,
                        tl.max_length,
                        tl.avg_length,
                    ))
                } else {
                    ColumnStats::DateTime(DateTimeStats::empty())
                }
            }
            DataType::Boolean => {
                let (true_count, false_count) = input.boolean_counts.unwrap_or_else(|| {
                    let tc = input
                        .sample_values
                        .iter()
                        .filter(|v| parse_strict_boolean_token(v.trim()) == Some(true))
                        .count();
                    let fc = input
                        .sample_values
                        .iter()
                        .filter(|v| parse_strict_boolean_token(v.trim()) == Some(false))
                        .count();
                    (tc, fc)
                });
                let total = true_count + false_count;
                if total > 0 {
                    ColumnStats::Boolean(BooleanStats {
                        true_count,
                        false_count,
                        true_ratio: true_count as f64 / total as f64,
                    })
                } else {
                    ColumnStats::None
                }
            }
            DataType::String | DataType::Identifier => {
                if let Some(tl) = &input.text_lengths {
                    ColumnStats::Text(TextStats::from_lengths(
                        tl.min_length,
                        tl.max_length,
                        tl.avg_length,
                    ))
                } else {
                    calculate_text_stats(input.sample_values)
                }
            }
            DataType::Nested => unreachable!("nested columns return above"),
        }
    };

    // A column with no non-null value has counts, but no statistics: every
    // aggregate over zero values is undefined, and the repo's convention for
    // undefined is absent, not zero. Without this guard, a column with no rows
    // reported a numeric min, max, mean, std_dev and variance of 0.0 and a
    // true_ratio of 0.0, and a text column of nulls reported lengths of 0
    // (#983): plausible numbers describing values that do not exist, which is
    // the failure class AGENTS.md singles out. 0.0 and 0 are ordinary values
    // for a real column, so nothing about that output looked wrong.
    //
    // The counts above are untouched. "0 of 0 values were invalid" is a fact
    // about a column that was analyzed, not a statistic over nothing.
    let stats = stats_over_values(stats, input.total_count, input.null_count);

    let patterns = if input.skip_patterns {
        None
    } else {
        Some(detect_patterns(input.sample_values, input.locale))
    };

    ColumnProfile {
        name: input.name,
        data_type: input.data_type,
        null_count: input.null_count,
        total_count: input.total_count,
        unique_count: input.unique_count,
        unique_count_is_approximate: input.unique_count_is_approximate,
        unique_count_lower_bound: input.unique_count_lower_bound,
        invalid_count,
        // Classified over the retained sample, so the counts are bounded by the
        // engine's reservoir on a large source. `classified_count()` against
        // `total_count - null_count` is what tells a reader which happened.
        // Not gated by `skip_statistics`: which forms a column holds is
        // schema-level evidence like the inferred type, not a statistic.
        type_homogeneity: Some(classify_lexical_forms(input.sample_values)),
        // Over the same retained sample, so it stays a subset of `text` above.
        locale_number_count: Some(count_locale_numbers(input.sample_values)),
        stats,
        patterns,
    }
}

/// Build the profile of a container column: structs, lists and maps.
///
/// It reports its counts and nothing else. Its lengths, distinct values,
/// patterns and lexical forms would all be measured on a serialisation, and
/// every input path serialises containers differently (#637). Absent is "not
/// analyzed", which is what they are.
pub fn nested_column_profile(name: String, total_count: usize, null_count: usize) -> ColumnProfile {
    ColumnProfile {
        name,
        data_type: DataType::Nested,
        null_count,
        total_count,
        unique_count: None,
        unique_count_is_approximate: None,
        unique_count_lower_bound: None,
        invalid_count: None,
        type_homogeneity: None,
        locale_number_count: None,
        stats: ColumnStats::None,
        patterns: None,
    }
}

/// Report every column whose non-null values were all containers as
/// [`DataType::Nested`], counts only (#637).
///
/// For text-based inputs that know which of their values were containers, as
/// the JSON readers do. `container_counts` holds, per column name, how many of
/// the analyzed values were an object or an array.
///
/// This matches what a typed source reports for the same records: Parquet and
/// Arrow carry a struct or list type, and profile such a column the same way.
/// A column that mixes containers with scalars has no typed twin, and stays
/// the text column its values make it.
///
/// The comparison runs against the profile's own counts, so a container is
/// weighed against exactly the values the null rule left standing. A container
/// always renders with a bracket, which the null rule never matches.
pub fn mark_container_columns(
    profiles: Vec<ColumnProfile>,
    container_counts: &HashMap<String, usize>,
) -> Vec<ColumnProfile> {
    profiles
        .into_iter()
        .map(|profile| {
            let containers = container_counts.get(&profile.name).copied().unwrap_or(0);
            let non_null = profile.total_count - profile.null_count;
            if containers > 0 && containers == non_null {
                nested_column_profile(profile.name, profile.total_count, profile.null_count)
            } else {
                profile
            }
        })
        .collect()
}

/// Convert all columns in a [`StreamingColumnCollection`] into [`ColumnProfile`]s.
pub fn profiles_from_streaming(
    column_stats: &StreamingColumnCollection,
    skip_statistics: bool,
    skip_patterns: bool,
    locale: Option<Locale>,
) -> Vec<ColumnProfile> {
    profiles_from_streaming_with_hints(
        column_stats,
        skip_statistics,
        skip_patterns,
        locale,
        &SemanticHints::default(),
    )
}

/// Convert all columns into [`ColumnProfile`]s while applying semantic hints.
pub fn profiles_from_streaming_with_hints(
    column_stats: &StreamingColumnCollection,
    skip_statistics: bool,
    skip_patterns: bool,
    locale: Option<Locale>,
    semantic_hints: &SemanticHints,
) -> Vec<ColumnProfile> {
    let mut profiles = Vec::new();

    for column_name in column_stats.column_names() {
        if let Some(stats) = column_stats.get_column_stats(&column_name) {
            let profile = profile_from_stats_with_hints(
                &column_name,
                stats,
                skip_statistics,
                skip_patterns,
                locale,
                semantic_hints,
            );
            profiles.push(profile);
        }
    }

    profiles
}

/// Convert a single column's [`StreamingStatistics`] into a [`ColumnProfile`].
pub fn profile_from_stats(
    name: &str,
    stats: &StreamingStatistics,
    skip_statistics: bool,
    skip_patterns: bool,
    locale: Option<Locale>,
) -> ColumnProfile {
    profile_from_stats_with_hints(
        name,
        stats,
        skip_statistics,
        skip_patterns,
        locale,
        &SemanticHints::default(),
    )
}

/// Convert a single column into a [`ColumnProfile`] while applying semantic hints.
pub fn profile_from_stats_with_hints(
    name: &str,
    stats: &StreamingStatistics,
    skip_statistics: bool,
    skip_patterns: bool,
    locale: Option<Locale>,
    semantic_hints: &SemanticHints,
) -> ColumnProfile {
    let data_type = if semantic_hints.is_identifier_column(name) {
        DataType::Identifier
    } else {
        infer_data_type_streaming(stats)
    };
    let text_stats = stats.text_length_stats();

    build_column_profile(ColumnProfileInput {
        name: name.to_string(),
        data_type,
        total_count: stats.count,
        null_count: stats.null_count,
        unique_count: Some(stats.unique_count()),
        unique_count_is_approximate: Some(stats.unique_count_is_approximate()),
        unique_count_lower_bound: Some(stats.unique_count_lower_bound()),
        sample_values: stats.sample_values(),
        text_lengths: Some(TextLengths {
            min_length: text_stats.min_length,
            max_length: text_stats.max_length,
            avg_length: text_stats.avg_length,
        }),
        boolean_counts: None,
        skip_statistics,
        skip_patterns,
        locale,
        exact_numeric: stats.exact_numeric_aggregates(),
        exact_date_matches: Some(stats.date_match_count()),
    })
}

/// Infer [`DataType`] from [`StreamingStatistics`] sample values.
pub fn infer_data_type_streaming(stats: &StreamingStatistics) -> DataType {
    let sample_values = stats.sample_values();
    let non_empty: Vec<&String> = sample_values
        .iter()
        .filter(|s| !is_null_like_token(s.trim()))
        .collect();

    if !non_empty.is_empty() {
        // The same question `infer_type` asks, so codes are text on every engine.
        let digit_codes = holds_digit_codes(non_empty.iter().map(|s| s.trim()));

        let all_integers = non_empty.iter().all(|s| is_integer_token(s.trim()));
        if !digit_codes && all_integers {
            return DataType::Integer;
        }

        // Numeric lexical forms remain numeric even if none are finite enough
        // to enter the streaming aggregates. `build_column_profile` reports
        // those unusable values through `invalid_count`.
        let numeric_count = non_empty
            .iter()
            .filter(|s| s.trim().parse::<f64>().is_ok())
            .count();
        if !digit_codes && numeric_count as f64 / non_empty.len() as f64 > 0.8 {
            return DataType::Float;
        }

        // `.trim()`, like every sibling predicate in this function and like
        // `infer_type`. Without it a value that arrived padded — CSV parsing
        // does not trim by default — was a date to the buffered engines and
        // text to this one, which is the same cross-engine split the shared
        // grammar exists to close. Over every value, as `infer_type` counts and
        // as the shares above do: a column whose first 100 values were dates
        // was a date here and text everywhere else (#915).
        let date_like_count = non_empty
            .iter()
            .filter(|s| is_inferred_date_token(s.trim()))
            .count();

        if date_like_count as f64 / non_empty.len() as f64 > 0.7 {
            return DataType::Date;
        }

        let bool_count = non_empty
            .iter()
            .filter(|s| parse_strict_boolean_token(s.trim()).is_some())
            .count();

        if bool_count as f64 / non_empty.len() as f64 >= 0.9 {
            return DataType::Boolean;
        }
    }

    DataType::String
}

/// Build a sample `HashMap` from a [`StreamingColumnCollection`] suitable for
/// `QualityMetrics::calculate_from_data()`.
pub fn quality_check_samples(
    column_stats: &StreamingColumnCollection,
) -> HashMap<String, Vec<String>> {
    let mut samples = HashMap::new();

    for column_name in column_stats.column_names() {
        if let Some(stats) = column_stats.get_column_stats(&column_name) {
            let sample_values: Vec<String> = stats.sample_values().to_vec();
            samples.insert(column_name, sample_values);
        }
    }

    samples
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::streaming_stats::StreamingColumnCollection;

    #[test]
    fn test_profiles_from_streaming() {
        let mut collection = StreamingColumnCollection::new();
        let headers = vec!["name".to_string(), "age".to_string()];

        collection.process_record(&headers, vec!["Alice".to_string(), "30".to_string()]);
        collection.process_record(&headers, vec!["Bob".to_string(), "25".to_string()]);
        collection.process_record(&headers, vec!["Charlie".to_string(), "35".to_string()]);

        let profiles = profiles_from_streaming(&collection, false, false, None);
        assert_eq!(profiles.len(), 2);

        let age = profiles.iter().find(|p| p.name == "age").unwrap();
        assert_eq!(age.data_type, DataType::Integer);
        assert_eq!(age.total_count, 3);
    }

    #[test]
    fn test_all_non_finite_tokens_keep_streaming_float_type() {
        let mut collection = StreamingColumnCollection::new();
        let headers = vec!["value".to_string()];

        collection.process_record(&headers, vec!["Infinity".to_string()]);
        collection.process_record(&headers, vec!["-inf".to_string()]);

        let profiles = profiles_from_streaming(&collection, false, false, None);
        let value = profiles
            .iter()
            .find(|profile| profile.name == "value")
            .expect("value profile");

        assert_eq!(value.data_type, DataType::Float);
        assert_eq!(value.invalid_count, Some(2));
        assert!(matches!(value.stats, ColumnStats::None));
    }

    #[test]
    fn type_homogeneity_is_classified_from_the_retained_sample() {
        // The streaming path only holds its reservoir, so the counts describe
        // the sample. `classified_count()` short of the non-null total is the
        // signal that the shares are sampled — nothing else discloses it.
        let samples = ["1", "2", "junk", ""].map(String::from).to_vec();
        let profile = build_column_profile(ColumnProfileInput {
            name: "v".to_string(),
            data_type: DataType::String,
            total_count: 4_000,
            null_count: 1_000,
            unique_count: Some(3),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            sample_values: &samples,
            text_lengths: None,
            boolean_counts: None,
            // Not gated by the analysis selection: a column's lexical makeup is
            // schema-level evidence, like its inferred type.
            skip_statistics: true,
            skip_patterns: true,
            exact_numeric: None,
            exact_date_matches: None,
            locale: None,
        });

        let counts = profile
            .type_homogeneity
            .expect("classification runs on every column");
        assert_eq!(counts.numeric, 2);
        assert_eq!(counts.text, 1);
        assert_eq!(counts.classified_count(), 3, "null-like values are skipped");
        assert!(
            counts.classified_count() < profile.total_count - profile.null_count,
            "a sample-derived count must read as short of the column"
        );
    }

    #[test]
    fn test_quality_check_samples() {
        let mut collection = StreamingColumnCollection::new();
        let headers = vec!["col".to_string()];

        collection.process_record(&headers, vec!["val1".to_string()]);
        collection.process_record(&headers, vec!["val2".to_string()]);

        let samples = quality_check_samples(&collection);
        assert!(samples.contains_key("col"));
        assert_eq!(samples["col"].len(), 2);
    }

    #[test]
    fn typed_columns_without_parsed_values_have_no_statistics() {
        for data_type in [DataType::Integer, DataType::Float, DataType::Boolean] {
            for values in [["", ""], ["junk", "Infinity"]] {
                let samples = values.map(String::from);
                let null_count = if values[0].is_empty() { 2 } else { 0 };
                let profile = build_column_profile(ColumnProfileInput {
                    name: "v".to_string(),
                    data_type: data_type.clone(),
                    total_count: 2,
                    null_count,
                    unique_count: None,
                    unique_count_is_approximate: None,
                    unique_count_lower_bound: None,
                    sample_values: &samples,
                    text_lengths: None,
                    boolean_counts: None,
                    skip_statistics: false,
                    skip_patterns: false,
                    locale: None,
                    exact_numeric: None,
                    exact_date_matches: None,
                });

                assert!(matches!(profile.stats, ColumnStats::None));
                assert_eq!(profile.data_type, data_type);
                assert_eq!(profile.total_count, 2);
                assert_eq!(profile.null_count, null_count);
                if matches!(data_type, DataType::Integer | DataType::Float) {
                    assert_eq!(profile.invalid_count, Some(2 - null_count));
                }
                assert!(profile.patterns.is_some());
            }
        }
    }

    #[test]
    fn exact_zero_statistics_survive_a_sample_without_parsed_values() {
        // A bounded sample can miss every usable value. The full-stream
        // count, not the sample count or the aggregate's value, governs absence.
        for data_type in [DataType::Integer, DataType::Boolean] {
            let samples = [String::new()];
            let profile = build_column_profile(ColumnProfileInput {
                name: "v".to_string(),
                data_type: data_type.clone(),
                total_count: 3,
                null_count: 1,
                unique_count: None,
                unique_count_is_approximate: None,
                unique_count_lower_bound: None,
                sample_values: &samples,
                text_lengths: None,
                boolean_counts: Some((0, 2)),
                skip_statistics: false,
                skip_patterns: true,
                locale: None,
                exact_numeric: Some(ExactNumericAggregates {
                    min: 0.0,
                    max: 0.0,
                    mean: 0.0,
                    std_dev: 0.0,
                    variance: 0.0,
                    count: 2,
                }),
                exact_date_matches: None,
            });

            match profile.stats {
                ColumnStats::Numeric(stats) => {
                    assert_eq!(stats.min, 0.0);
                    assert_eq!(stats.mean, 0.0);
                    assert_eq!(stats.is_approximate, Some(true));
                    assert_eq!(profile.invalid_count, Some(0));
                }
                ColumnStats::Boolean(stats) => {
                    assert_eq!(stats.true_ratio, 0.0);
                    assert_eq!(stats.false_count, 2);
                }
                other => panic!("expected measured zero statistics, got {other:?}"),
            }
        }
    }

    #[test]
    fn an_overflowing_exact_spread_clears_the_sample_shape() {
        // The bounded sample missed the extreme values, so on its own it has a
        // finite spread; the full stream does not (#784).
        let samples: Vec<String> = ["5", "0.5", "5", "7", "3"]
            .iter()
            .map(|value| value.to_string())
            .collect();
        let profile = build_column_profile(ColumnProfileInput {
            name: "x".to_string(),
            data_type: DataType::Float,
            total_count: 20_000,
            null_count: 0,
            unique_count: None,
            unique_count_is_approximate: None,
            unique_count_lower_bound: None,
            sample_values: &samples,
            text_lengths: None,
            boolean_counts: None,
            skip_statistics: false,
            skip_patterns: true,
            locale: None,
            exact_numeric: Some(ExactNumericAggregates {
                min: -1e308,
                max: 1e308,
                mean: 2.9167,
                std_dev: f64::NAN,
                variance: f64::NAN,
                count: 20_000,
            }),
            exact_date_matches: None,
        });

        let ColumnStats::Numeric(stats) = profile.stats else {
            panic!("expected numeric statistics, got {:?}", profile.stats);
        };
        assert_eq!(stats.variance, None);
        assert_eq!(stats.std_dev, None);
        assert_eq!(stats.coefficient_of_variation, None);
        assert_eq!(stats.skewness, None);
        assert_eq!(stats.kurtosis, None);
        // Order statistics still describe the retained sample.
        assert_eq!(stats.median, Some(5.0));
        assert_eq!(stats.is_approximate, Some(true));
    }

    #[test]
    fn test_boolean_stats_with_counts() {
        let samples = vec!["True".to_string(), "False".to_string(), "True".to_string()];
        let profile = build_column_profile(ColumnProfileInput {
            name: "flag".to_string(),
            data_type: DataType::Boolean,
            total_count: 3,
            null_count: 0,
            unique_count: Some(2),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            sample_values: &samples,
            text_lengths: None,
            boolean_counts: Some((2, 1)),
            skip_statistics: false,
            skip_patterns: false,
            exact_numeric: None,
            exact_date_matches: None,
            locale: None,
        });

        match &profile.stats {
            ColumnStats::Boolean(b) => {
                assert_eq!(b.true_count, 2);
                assert_eq!(b.false_count, 1);
                assert!((b.true_ratio - 2.0 / 3.0).abs() < 0.001);
            }
            other => panic!("expected Boolean stats, got {:?}", other),
        }
    }

    #[test]
    fn test_boolean_stats_fallback_case_insensitive() {
        let samples = vec![
            "true".to_string(),
            "FALSE".to_string(),
            " True ".to_string(),
        ];
        let profile = build_column_profile(ColumnProfileInput {
            name: "flag".to_string(),
            data_type: DataType::Boolean,
            total_count: 3,
            null_count: 0,
            unique_count: Some(2),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            sample_values: &samples,
            text_lengths: None,
            boolean_counts: None,
            skip_statistics: false,
            skip_patterns: false,
            exact_numeric: None,
            exact_date_matches: None,
            locale: None,
        });

        match &profile.stats {
            ColumnStats::Boolean(b) => {
                assert_eq!(b.true_count, 2);
                assert_eq!(b.false_count, 1);
                assert!((b.true_ratio - 2.0 / 3.0).abs() < 0.001);
            }
            other => panic!("expected Boolean stats, got {:?}", other),
        }
    }

    #[test]
    fn test_skip_statistics() {
        let samples = vec!["10".to_string(), "20".to_string(), "30".to_string()];
        let profile = build_column_profile(ColumnProfileInput {
            name: "num".to_string(),
            data_type: DataType::Integer,
            total_count: 3,
            null_count: 0,
            unique_count: Some(3),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            sample_values: &samples,
            text_lengths: None,
            boolean_counts: None,
            skip_statistics: true,
            skip_patterns: false,
            exact_numeric: None,
            exact_date_matches: None,
            locale: None,
        });

        assert!(matches!(profile.stats, ColumnStats::None));
        assert_eq!(profile.data_type, DataType::Integer);
    }

    #[test]
    fn test_skip_patterns() {
        let samples = vec!["hello".to_string(), "world".to_string()];
        let profile = build_column_profile(ColumnProfileInput {
            name: "text".to_string(),
            data_type: DataType::String,
            total_count: 2,
            null_count: 0,
            unique_count: Some(2),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            sample_values: &samples,
            text_lengths: None,
            boolean_counts: None,
            skip_statistics: false,
            skip_patterns: true,
            exact_numeric: None,
            exact_date_matches: None,
            locale: None,
        });

        // skip_patterns must be distinguishable from "scanned, nothing matched",
        // otherwise downstream redaction gates cannot fail closed.
        assert!(profile.patterns.is_none());
        assert!(matches!(profile.stats, ColumnStats::Text(_)));
    }

    #[test]
    fn test_scanned_without_matches_is_not_none() {
        // The counterpart to `test_skip_patterns`: a column that *was* scanned
        // and matched nothing yields `Some([])`, which downstream consumers may
        // safely read as "no sensitive data here".
        let samples = vec!["hello".to_string(), "world".to_string()];
        let profile = build_column_profile(ColumnProfileInput {
            name: "text".to_string(),
            data_type: DataType::String,
            total_count: 2,
            null_count: 0,
            unique_count: Some(2),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            sample_values: &samples,
            text_lengths: None,
            boolean_counts: None,
            skip_statistics: false,
            skip_patterns: false,
            exact_numeric: None,
            exact_date_matches: None,
            locale: None,
        });

        assert!(profile.patterns.is_some_and(|p| p.is_empty()));
    }

    #[test]
    fn test_all_packs_default() {
        let samples = vec!["42".to_string(), "99".to_string()];
        let profile = build_column_profile(ColumnProfileInput {
            name: "val".to_string(),
            data_type: DataType::Integer,
            total_count: 2,
            null_count: 0,
            unique_count: Some(2),
            unique_count_is_approximate: Some(false),
            unique_count_lower_bound: None,
            sample_values: &samples,
            text_lengths: None,
            boolean_counts: None,
            skip_statistics: false,
            skip_patterns: false,
            exact_numeric: None,
            exact_date_matches: None,
            locale: None,
        });

        assert!(matches!(profile.stats, ColumnStats::Numeric(_)));
        assert_eq!(profile.data_type, DataType::Integer);
    }
}