moarstats.rs

static USAGE: &str = r#"
Add dozens of additional statistics, including extended outlier, robust & bivariate
statistics to an existing stats CSV file. It also maps the field type to the most specific
W3C XML Schema Definition (XSD) datatype (https://www.w3.org/TR/xmlschema-2/).

The `moarstats` command extends an existing stats CSV file (created by the `stats` command)
by computing "moar" (https://www.dictionary.com/culture/slang/moar) statistics that can be
derived from existing stats columns and by scanning the original CSV file.

It looks for the `<FILESTEM>.stats.csv` file for a given CSV input. If the stats CSV file
does not exist, it will first run the `stats` command with configurable options to establish
the baseline stats, to which it will add more stats columns.

If the `.stats.csv` file is found, it will skip running stats and just append the additional
stats columns.

Currently computes the following 25 additional univariate statistics:
 1. Pearson's Second Skewness Coefficient: 3 * (mean - median) / stddev
    Measures asymmetry of the distribution.
    Positive values indicate right skew, negative values indicate left skew.
    https://en.wikipedia.org/wiki/Skewness
 2. Range to Standard Deviation Ratio: range / stddev
    Normalizes the spread of data.
    Higher values indicate more extreme outliers relative to the variability.
 3. Quartile Coefficient of Dispersion: (Q3 - Q1) / (Q3 + Q1)
    Measures relative variability using quartiles.
    Useful for comparing dispersion across different scales.
    https://en.wikipedia.org/wiki/Quartile_coefficient_of_dispersion
 4. Z-Score of Mode: (mode - mean) / stddev
    Indicates how typical the mode is relative to the distribution.
    Values near 0 suggest the mode is near the mean.
 5. Relative Standard Error: sem / mean
    Measures precision of the mean estimate relative to its magnitude.
    Lower values indicate more reliable estimates.
 6. Z-Score of Min: (min - mean) / stddev
    Shows how extreme the minimum value is.
    Large negative values indicate outliers or heavy left tail.
 7. Z-Score of Max: (max - mean) / stddev
    Shows how extreme the maximum value is.
    Large positive values indicate outliers or heavy right tail.
 8. Median-to-Mean Ratio: median / mean
    Indicates skewness direction.
    Ratio < 1 suggests right skew, > 1 suggests left skew, = 1 suggests symmetry.
 9. IQR-to-Range Ratio: iqr / range
    Measures concentration of data.
    Higher values (closer to 1) indicate more data concentrated in the middle 50%.
10. MAD-to-StdDev Ratio: mad / stddev
    Compares robust vs non-robust spread measures.
    Higher values suggest presence of outliers affecting stddev.
11. Trimean: (Q1 + 2*median + Q3) / 4
    Tukey's trimean - a robust estimator of central tendency combining the median
    with the midhinge. More robust than mean, more efficient than median alone.
    https://en.wikipedia.org/wiki/Trimean
12. Midhinge: (Q1 + Q3) / 2
    Midpoint of the middle 50% of data. A robust central tendency measure
    that complements the mean and median.
    https://en.wikipedia.org/wiki/Midhinge
13. Robust CV: MAD / |median|
    Robust Coefficient of Variation using MAD and the magnitude of the median.
    Always non-negative. Resistant to outliers, useful for comparing variability.
    https://en.wikipedia.org/wiki/Robust_measures_of_scale
14. Kurtosis: Measures the "tailedness" of the distribution (excess kurtosis).
    Positive values indicate heavy tails, negative values indicate light tails.
    Values near 0 indicate a normal distribution.
    Requires --advanced flag.
    https://en.wikipedia.org/wiki/Kurtosis
15. Bimodality Coefficient: Measures whether a distribution has two modes (peaks) or is unimodal.
    BC < 0.555 indicates unimodal, BC >= 0.555 indicates bimodal/multimodal.
    Computed as (skewness² + 1) / (kurtosis + 3).
    Requires --advanced flag (needs skewness from base stats and kurtosis from --advanced flag).
    https://en.wikipedia.org/wiki/Bimodality
16. Jarque-Bera Test: (n/6) * (S² + K²/4)
    Standard test for normality using skewness and kurtosis.
    Also computes jarque_bera_pvalue (from chi-squared distribution with 2 df).
    Low p-values (< 0.05) indicate the data is NOT normally distributed.
    Requires --advanced flag (needs kurtosis).
    https://en.wikipedia.org/wiki/Jarque%E2%80%93Bera_test
17. Gini Coefficient: Measures inequality/dispersion in the distribution.
    Values range from 0 (perfect equality) to 1 (maximum inequality).
    Requires --advanced flag.
    https://en.wikipedia.org/wiki/Gini_coefficient
18. Atkinson Index: Measures inequality in the distribution with a sensitivity parameter.
    Values range from 0 (perfect equality) to 1 (maximum inequality).
    The Atkinson Index is a more general form of the Gini coefficient that allows for
    different sensitivity to inequality. Sensitivity is configurable via --epsilon.
    Requires --advanced flag.
    https://en.wikipedia.org/wiki/Atkinson_index
19. Theil Index: (1/n) * Σ((x_i / mean) * ln(x_i / mean))
    Measures inequality/concentration. Unlike Gini, it is decomposable into
    within-group and between-group components. Only computed for positive values.
    Requires --advanced flag.
    https://en.wikipedia.org/wiki/Theil_index
20. Mean Absolute Deviation (from mean): (1/n) * Σ|x_i - mean|
    Average absolute distance from the mean. Different from MAD (which uses median).
    Less robust but more statistically efficient than MAD.
    Requires --advanced flag.
21. Shannon Entropy: Measures the information content/uncertainty in the distribution.
    Higher values indicate more diversity, lower values indicate more concentration.
    Values range from 0 (all values identical) to log2(n) where n is the number of unique values.
    Requires --advanced flag.
    https://en.wikipedia.org/wiki/Entropy_(information_theory)
22. Normalized Entropy: Normalized version of Shannon Entropy scaled to [0, 1].
    Values range from 0 (all values identical) to 1 (all values equally distributed).
    Computed as shannon_entropy / log2(cardinality).
    Requires shannon_entropy (from --advanced flag) and cardinality (from base stats).
23. Simpson's Diversity Index: 1 - Σ(p_i²)
    Probability that two randomly chosen values are different.
    Ranges from 0 (all identical) to 1 (all unique). More intuitive than entropy.
    Requires --advanced flag (computed alongside entropy from frequency data).
    https://en.wikipedia.org/wiki/Diversity_index#Simpson_index
24. Winsorized Mean: Replaces values below/above thresholds with threshold values, then computes mean.
    All values are included in the calculation, but extreme values are capped at thresholds.
    https://en.wikipedia.org/wiki/Winsorized_mean
    Also computes: winsorized_stddev, winsorized_variance, winsorized_cv, winsorized_range,
    and winsorized_stddev_ratio (winsorized_stddev / overall_stddev).
25. Trimmed Mean: Excludes values outside thresholds, then computes mean.
    Only values within thresholds are included in the calculation.
    https://en.wikipedia.org/wiki/Truncated_mean
    Also computes: trimmed_stddev, trimmed_variance, trimmed_cv, trimmed_range,
    and trimmed_stddev_ratio (trimmed_stddev / overall_stddev).
    By default, uses Q1 and Q3 as thresholds (25% winsorization/trimming).
    With --use-percentiles, uses configurable percentiles (e.g., 5th/95th) as thresholds
    with --pct-thresholds.

In addition, it computes the following univariate outlier statistics (24 outlier statistics total).
https://en.wikipedia.org/wiki/Outlier
(requires --quartiles or --everything in stats):

Outlier Counts (7 statistics):
  - outliers_extreme_lower_cnt: Count of values below the lower outer fence
  - outliers_mild_lower_cnt: Count of values between lower outer and inner fences
  - outliers_normal_cnt: Count of values between inner fences (non-outliers)
  - outliers_mild_upper_cnt: Count of values between upper inner and outer fences
  - outliers_extreme_upper_cnt: Count of values above the upper outer fence
  - outliers_total_cnt: Total count of all outliers (sum of extreme and mild outliers)
  - outliers_percentage: Percentage of values that are outliers

Outlier Descriptive Statistics (6 statistics):
  - outliers_mean: Mean value of outliers
  - non_outliers_mean: Mean value of non-outliers
  - outliers_to_normal_mean_ratio: Ratio of outlier mean to non-outlier mean
  - outliers_min: Minimum value among outliers
  - outliers_max: Maximum value among outliers
  - outliers_range: Range of outlier values (max - min)

Outlier Variance/Spread Statistics (7 statistics):
  - outliers_stddev: Standard deviation of outlier values
  - outliers_variance: Variance of outlier values
  - non_outliers_stddev: Standard deviation of non-outlier values
  - non_outliers_variance: Variance of non-outlier values
  - outliers_cv: Coefficient of variation for outliers (stddev / mean)
  - non_outliers_cv: Coefficient of variation for non-outliers (stddev / mean)
  - outliers_normal_stddev_ratio: Ratio of outlier stddev to non-outlier stddev

Outlier Impact Statistics (2 statistics):
  - outlier_impact: Difference between overall mean and non-outlier mean
  - outlier_impact_ratio: Relative impact (outlier_impact / non_outlier_mean)

Outlier Boundary Statistics (2 statistics):
  - lower_outer_fence_zscore: Z-score of the lower outer fence boundary
  - upper_outer_fence_zscore: Z-score of the upper outer fence boundary

  These outlier statistics require reading the original CSV file and comparing each
  value against the fence thresholds.
  Fences are computed using the IQR method:
    inner fences at Q1/Q3 ± 1.5*IQR, outer fences at Q1/Q3 ± 3.0*IQR.

These univariate statistics are only computed for numeric and date/datetime columns
where the required base univariate statistics (mean, median, stddev, etc.) are available.
Univariate outlier statistics additionally require that quartiles (and thus fences) were
computed when generating the stats CSV.
Winsorized/trimmed means require either Q1/Q3 or percentiles to be available.
Kurtosis, Gini & Atkinson Index require reading the original CSV file to collect
all values for computation.

BIVARIATE STATISTICS:

The `moarstats` command also computes the following 6 bivariate statistics:
 1. Pearson's correlation
    Measures linear correlation between two numeric/date fields.
    Values range from -1 (perfect negative correlation) to +1 (perfect positive correlation).
    0 indicates no linear correlation.
    https://en.wikipedia.org/wiki/Pearson_correlation_coefficient
 2. Spearman's rank correlation
    Measures monotonic correlation between two numeric/date fields.
    Values range from -1 (perfect negative correlation) to +1 (perfect positive correlation).
    0 indicates no monotonic correlation.
    https://en.wikipedia.org/wiki/Spearman%27s_rank_correlation_coefficient
 3. Kendall's tau
    Measures monotonic correlation between two numeric/date fields.
    Values range from -1 (perfect negative correlation) to +1 (perfect positive correlation).
    0 indicates no monotonic correlation.
    https://en.wikipedia.org/wiki/Kendall_rank_correlation_coefficient
 4. Covariance
    Measures the linear relationship between two numeric/date fields.
    Values range from negative infinity to positive infinity.
    0 indicates no linear relationship.
    https://en.wikipedia.org/wiki/Covariance
 5. Mutual Information
    Measures the amount of information obtained about one field by observing another.
    Values range from 0 (independent) to positive infinity.
    https://en.wikipedia.org/wiki/Mutual_information
 6. Normalized Mutual Information
    Normalized version of mutual information, scaled by the geometric mean of individual entropies.
    Values range from 0 (independent) to 1 (perfectly dependent).
    https://en.wikipedia.org/wiki/Mutual_information#Normalized_variants

These bivariate statistics are computed when the `--bivariate` flag is used
and require an indexed CSV file (index will be auto-created if missing).
Bivariate statistics are output to a separate file: `<FILESTEM>.stats.bivariate.csv`.

Bivariate statistics require reading the entire CSV file and are computationally VERY expensive.
For large files (>= 10k records), parallel chunked processing is used when an index is available.
For smaller files or when no index exists, sequential processing is used.

MULTI-DATASET BIVARIATE STATISTICS:

When using the `--join-inputs` flag, multiple datasets can be joined internally before
computing bivariate statistics. This allows analyzing bivariate statistics across datasets
that share common join keys. The joined dataset is saved as a temporary file that is
automatically deleted after computing the bivariate statistics.
The bivariate statistics are saved to `<FILESTEM>.stats.bivariate.joined.csv`.

Non-finite numeric tokens ("NaN", "Infinity", "-Infinity", and their case variants) are
excluded from moarstats computations — the parser in moarstats filters them out before they
reach correlation, variance and mean calculations, preventing a single bad cell from silently
poisoning the results. Note that the baseline `stats` command may still count these tokens
as Float observations, so the `type`/`null_count` columns in `<FILESTEM>.stats.csv` are not
affected by this filter.

Examples:

  # Add moar stats to existing stats file
  qsv moarstats data.csv

  # Generate baseline stats first with custom options, then add moar stats
  qsv moarstats data.csv --stats-options "--everything --infer-dates"

  # Compute bivariate statistics between fields
  qsv moarstats data.csv --bivariate

  # Compute even more bivariate statistics
  qsv moarstats data.csv --bivariate --bivariate-stats pearson,spearman,kendall,mi,nmi,covariance

  # Join multiple datasets and compute bivariate statistics
  qsv moarstats data.csv --bivariate --join-inputs customers.csv,products.csv --join-keys cust_id,prod_id

  # Join multiple datasets and compute bivariate statistics with different join type
  qsv moarstats data.csv --bivariate --join-inputs customers.csv,products.csv --join-keys cust_id,prod_id --join-type left

For more examples, see https://github.com/dathere/qsv/blob/master/tests/test_moarstats.rs.

Usage:
    qsv moarstats [options] [<input>]
    qsv moarstats --help

moarstats options:
    --advanced             Compute Kurtosis, Shannon Entropy, Bimodality Coefficient,
                           Jarque-Bera, Gini Coefficient, Atkinson Index, Theil Index,
                           Mean Absolute Deviation, and Simpson's Diversity Index.
                           These advanced statistics computations require reading the
                           original CSV file to collect all values
                           for computation and are computationally expensive.
                           Further, Entropy computation requires the frequency command
                           to be run with --limit 0 to collect all frequencies.
                           An index will be auto-created for the original CSV file
                           if it doesn't already exist to enable parallel processing.
    -e, --epsilon <n>      The Atkinson Index Inequality Aversion parameter.
                           Epsilon controls the sensitivity of the Atkinson Index to inequality.
                           The higher the epsilon, the more sensitive the index is to inequality.
                           Typical values are 0.5 (standard in economic research),
                           1.0 (natural boundary), or 2.0 (useful for poverty analysis).
                           [default: 1.0]
    --stats-options <arg>  Options to pass to the stats command if baseline stats need
                           to be generated. The options are passed as a single string
                           that will be split by whitespace.
                           [default: --infer-dates --infer-boolean --mad --quartiles --percentiles --force --stats-jsonl]
    --round <n>            Round statistics to <n> decimal places. Rounding follows
                           Midpoint Nearest Even (Bankers Rounding) rule.
                           [default: 4]
    --use-percentiles      Use percentiles instead of Q1/Q3 for winsorization/trimming.
                           Requires percentiles to be computed in the stats CSV.
   --pct-thresholds <arg>  Comma-separated percentile pair (e.g., "10,90") to use
                           for winsorization/trimming when --use-percentiles is set.
                           Both values must be between 0 and 100, and lower < upper.
                           [default: 5,95]
 --xsd-gdate-scan <mode>   Gregorian XSD date type detection mode.
                           "quick": Fast detection using min/max values.
                                    Produces types with ?? suffix (less confident).
                           "thorough": Comprehensive detection checking all percentile values.
                                     Slower but ensures all values match the pattern.
                                     Produces types with ? suffix (more confident).
                           [default: quick]

                           BIVARIATE STATISTICS OPTIONS:
    -B, --bivariate        Enable bivariate statistics computation.
                           Requires indexed CSV file (index will be auto-created if missing).
                           Computes pairwise correlations, covariances, mutual information, and
                           normalized mutual information between columns. The bivariate statistics
                           are saved to a separate file in the same directory as the input:
                           <FILESTEM>.stats.bivariate.csv.
    -S, --bivariate-stats <stats>
                           Comma-separated list of bivariate statistics to compute.
                           Options: pearson, spearman, kendall, covariance, mi (mutual information),
                           nmi (normalized mutual information)
                           Use "all" to compute all statistics or "fast" to compute only
                           pearson & covariance, which is much faster as it doesn't require storing
                           all values and uses streaming algorithms.
                           [default: fast]
    -C, --cardinality-threshold <n>
                           Skip mutual information computation for field pairs where either
                           field has cardinality exceeding this threshold. Helps avoid
                           expensive computations for high-cardinality fields.
                           [default: 1000000]
    -J, --join-inputs <files>
                           Additional datasets to join. Comma-separated list of CSV files to join
                           with the primary input.
                           e.g.: --join-inputs customers.csv,products.csv
    -K, --join-keys <keys>
                           Join keys for each dataset. Comma-separated list of join key column names,
                           one per dataset. Must specify same number of keys as datasets (primary + addl).
                           e.g.: --join-keys customer_id,customer_id,product_id
    -T, --join-type <type>
                           Join type when using --join-inputs.
                           Valid values: inner, left, right, full
                           [default: inner]
    -p, --progressbar      Show progress bars when computing bivariate statistics.

Common options:
    --force                Force recomputing stats even if valid precomputed stats
                           cache exists.
    -j, --jobs <arg>       The number of jobs to run in parallel.
                           This works only when the given CSV has an index.
                           Note that a file handle is opened for each job.
                           When not set, the number of jobs is set to the
                           number of CPUs detected.
    -h, --help             Display this message
    -o, --output <file>    Write output to <file> instead of overwriting the stats CSV file.
"#;

use std::{
    env, fs,
    path::{Path, PathBuf},
    process::Command,
    sync::Arc,
    time::Instant,
};

use crossbeam_channel;
use csv::{ReaderBuilder, StringRecord, WriterBuilder};
use foldhash::{HashMap, HashMapExt, HashSet, HashSetExt};
use indexmap::IndexMap;
use indicatif::{HumanCount, ProgressBar, ProgressDrawTarget, ProgressStyle};
use qsv_dateparser::parse_with_preference;
use rayon::prelude::*;
use serde::Deserialize;
use simdutf8::basic::from_utf8;
use stats::{atkinson, gini, kurtosis};
use threadpool::ThreadPool;

use crate::{CliError, CliResult, config::Config, regex_oncelock, util};

/// Minimum record count before parallel processing is worthwhile for outliers and
/// bivariate stats. Below this, the scheduling overhead outweighs the speedup, so
/// fall back to the sequential path.
const PARALLEL_THRESHOLD: usize = 10_000;

#[derive(Debug, Deserialize)]
struct Args {
    arg_input:                  Option<String>,
    flag_stats_options:         String,
    flag_round:                 u32,
    flag_output:                Option<String>,
    flag_use_percentiles:       bool,
    flag_pct_thresholds:        Option<String>,
    flag_xsd_gdate_scan:        Option<String>,
    flag_advanced:              bool,
    flag_epsilon:               f64,
    flag_bivariate:             bool,
    flag_bivariate_stats:       String,
    flag_cardinality_threshold: Option<u64>,
    flag_join_inputs:           Option<String>,
    flag_join_keys:             Option<String>,
    flag_join_type:             Option<String>,
    flag_progressbar:           bool,
    flag_jobs:                  Option<usize>,
    flag_force:                 bool,
}

/// Configuration for which bivariate statistics to compute
#[derive(Clone, Copy, Debug, Default)]
struct BivariateStatsConfig {
    pearson:    bool,
    spearman:   bool,
    kendall:    bool,
    covariance: bool,
    mi:         bool, // mutual information
    nmi:        bool, // normalized mutual information
}

impl BivariateStatsConfig {
    /// Parse the --bivariate-stats flag value
    fn from_flag(flag_value: &str) -> CliResult<Self> {
        let mut config = Self::default();
        let mut invalid_stats = Vec::new();

        let flag_lower = flag_value.to_lowercase();
        for stat in flag_lower.split(',') {
            let stat_trimmed = stat.trim();
            if stat_trimmed.is_empty() {
                continue; // Skip empty entries from trailing commas
            }
            match stat_trimmed {
                "pearson" => config.pearson = true,
                "spearman" => config.spearman = true,
                "kendall" => config.kendall = true,
                "covariance" | "cov" => config.covariance = true,
                "mi" | "mutual_information" | "mutual-information" => config.mi = true,
                "nmi" | "normalized_mutual_information" | "normalized-mutual-information" => {
                    config.nmi = true;
                },
                "all" => return Ok(Self::all()),
                "fast" => {
                    config.pearson = true;
                    config.covariance = true;
                },
                _ => {
                    invalid_stats.push(stat_trimmed.to_string());
                },
            }
        }

        if !invalid_stats.is_empty() {
            return fail_clierror!(
                "Invalid bivariate statistics: {}. Valid options are: pearson, spearman, kendall, \
                 covariance (or cov), mi (or mutual_information or mutual-information), nmi (or \
                 normalized_mutual_information or normalized-mutual-information), fast, all",
                invalid_stats.join(", ")
            );
        }

        // Check if at least one stat was requested
        if !config.pearson
            && !config.spearman
            && !config.kendall
            && !config.covariance
            && !config.mi
            && !config.nmi
        {
            return fail_clierror!(
                "No valid bivariate statistics specified. Valid options are: pearson, spearman, \
                 kendall, covariance (or cov), mi (or mutual_information or mutual-information), \
                 nmi (or normalized_mutual_information or normalized-mutual-information), fast, \
                 all"
            );
        }

        Ok(config)
    }

    /// Enable all bivariate statistics
    const fn all() -> Self {
        Self {
            pearson:    true,
            spearman:   true,
            kendall:    true,
            covariance: true,
            mi:         true,
            nmi:        true,
        }
    }

    /// Check if we need to store all values (required for Spearman/Kendall)
    #[inline]
    const fn needs_all_values(self) -> bool {
        self.spearman || self.kendall
    }

    /// Check if we need frequency counts (required for mutual information and normalized mutual
    /// information)
    #[inline]
    const fn needs_frequency_counts(self) -> bool {
        self.mi || self.nmi
    }
}

/// Get the absolute stats CSV file path for a given input CSV path.
/// Delegates to the shared implementation in `util::get_stats_csv_path`.
fn get_stats_csv_path(input_path: &Path) -> CliResult<PathBuf> {
    util::get_stats_csv_path(input_path)
}

/// Get the absolute bivariate CSV file path for a given input CSV path
/// If `is_joined` is true, appends `.joined` to the filename before `.csv`
fn get_bivariate_csv_path(input_path: &Path, is_joined: bool) -> CliResult<PathBuf> {
    let parent = input_path.parent().unwrap_or_else(|| Path::new("."));
    let fstem = input_path
        .file_stem()
        .ok_or_else(|| CliError::Other("Invalid input path: no file name".to_string()))?;

    let bivariate_filename = if is_joined {
        format!("{}.stats.bivariate.joined.csv", fstem.to_string_lossy())
    } else {
        format!("{}.stats.bivariate.csv", fstem.to_string_lossy())
    };
    let result = parent.join(bivariate_filename);
    if result.is_absolute() {
        Ok(result)
    } else {
        Ok(std::env::current_dir()?.join(result))
    }
}

/// Join multiple datasets internally using join
fn join_datasets_internal(
    primary_input: &Path,
    additional_inputs: &[String],
    join_keys: &[String],
    join_type: &str,
) -> CliResult<PathBuf> {
    use tempfile::NamedTempFile;

    if additional_inputs.is_empty() {
        return fail_clierror!("No additional datasets provided for joining");
    }

    if join_keys.len() != additional_inputs.len() + 1 {
        return fail_clierror!(
            "Number of join keys ({}) must match number of datasets ({})",
            join_keys.len(),
            additional_inputs.len() + 1
        );
    }

    // Create temporary file for joined output with .csv extension
    let temp_dir =
        crate::config::TEMP_FILE_DIR.get_or_init(|| tempfile::TempDir::new().unwrap().keep());
    let temp_file = tempfile::Builder::new()
        .suffix(".csv")
        .tempfile_in(temp_dir)?;
    let temp_path = temp_file.path().to_path_buf();
    drop(temp_file); // Close the file so join can write to it

    let temp_path_str = temp_path
        .to_str()
        .ok_or_else(|| CliError::Other("Invalid temp path".to_string()))?
        .to_string();

    let primary_input_str = primary_input
        .to_str()
        .ok_or_else(|| CliError::Other("Invalid input path".to_string()))?
        .to_string();

    // Build join command arguments
    let join_type_flag: Option<&str> = match join_type {
        "left" => Some("--left"),
        "right" => Some("--right"),
        "full" => Some("--full"),
        _ => None, // inner is default
    };

    // Join datasets sequentially (join first additional to primary, then next to result, etc.)
    // This is simpler than handling multiple joins at once
    let mut current_input = primary_input_str;
    let mut current_key = join_keys[0].clone();

    // Resolve the executable path once before the loop
    let qsv_path = env::current_exe()
        .map_err(|e| CliError::Other(format!("Failed to get current executable path: {e:?}")))?
        .to_string_lossy()
        .to_string();

    // These are never read, but we need to declare them to avoid compiler errors
    #[allow(clippy::collection_is_never_read)]
    let mut intermediate_temps: Vec<NamedTempFile> = Vec::new();
    #[allow(clippy::collection_is_never_read)]
    let mut intermediate_path_strs: Vec<String> = Vec::new();

    for (i, (additional_input, next_key)) in additional_inputs
        .iter()
        .zip(join_keys[1..].iter())
        .enumerate()
    {
        let mut args: Vec<&str> = Vec::new();

        // Add join type flag if specified
        if let Some(flag) = join_type_flag {
            args.push(flag);
        }

        args.push(&current_key);
        args.push(&current_input);
        args.push(next_key);
        args.push(additional_input);

        let output_path_str = if i == additional_inputs.len() - 1 {
            // Last join - use final temp path
            temp_path_str.clone()
        } else {
            // Intermediate join - create another temp file with .csv extension
            let intermediate_temp = tempfile::Builder::new()
                .suffix(".csv")
                .tempfile_in(temp_dir)?;
            let intermediate_path = intermediate_temp.path().to_path_buf();
            intermediate_temps.push(intermediate_temp); // Keep temp file alive
            let intermediate_path_str = intermediate_path
                .to_str()
                .ok_or_else(|| CliError::Other("Invalid intermediate temp path".to_string()))?
                .to_string();
            intermediate_path_strs.push(intermediate_path_str.clone());
            intermediate_path_str
        };
        args.push("--output");
        args.push(&output_path_str);

        // Construct join command directly since it doesn't fit run_qsv_cmd pattern
        // (join takes two input files, not one)
        let mut cmd = Command::new(&qsv_path);
        cmd.arg("join").args(&args);

        let output = cmd
            .output()
            .map_err(|e| CliError::Other(format!("Error while executing join command: {e:?}")))?;

        if !output.status.success() {
            return fail_clierror!(
                "Command join failed: Output {{ status: {:?}, stdout: {:?}, stderr: {:?} }}",
                output.status,
                String::from_utf8_lossy(&output.stdout),
                String::from_utf8_lossy(&output.stderr)
            );
        }

        log::info!("Joining datasets...");

        // Update for next iteration
        current_input = output_path_str;
        current_key.clone_from(next_key);
    }

    Ok(temp_path)
}

/// Compute Pearson's Second Skewness Coefficient: 3 * (mean - median) / stddev
#[inline]
fn compute_pearson_skewness(
    mean: Option<f64>,
    median: Option<f64>,
    stddev: Option<f64>,
) -> Option<f64> {
    if let (Some(mean_val), Some(median_val), Some(stddev_val)) = (mean, median, stddev) {
        if stddev_val.abs() > f64::EPSILON {
            Some(3.0 * (mean_val - median_val) / stddev_val)
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute Range to Standard Deviation Ratio: range / stddev
#[inline]
fn compute_range_stddev_ratio(range: Option<f64>, stddev: Option<f64>) -> Option<f64> {
    if let (Some(range_val), Some(stddev_val)) = (range, stddev) {
        if stddev_val.abs() > f64::EPSILON {
            Some(range_val / stddev_val)
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute Quartile Coefficient of Dispersion: (Q3 - Q1) / (Q3 + Q1)
///
/// Note: If Q1 or Q3 are negative, especially if both are negative and equal in magnitude,
/// the denominator (Q3 + Q1) may be zero or near zero, causing the result to be `None`.
/// Also, the standard formula may not yield meaningful results if Q1 is negative and
/// Q1 >= Q3 (i.e., quartiles are not in the expected order).
/// Return None if quartiles are not in a valid order (Q1 < Q3), or denominator is 0.
#[inline]
fn compute_quartile_coefficient_dispersion(q1: Option<f64>, q3: Option<f64>) -> Option<f64> {
    if let (Some(q1_val), Some(q3_val)) = (q1, q3) {
        // Check that quartile order is valid (Q1 < Q3)
        if q1_val >= q3_val {
            return None;
        }
        let sum = q3_val + q1_val;
        // Only compute if the denominator is effectively non-zero to avoid division by zero and
        // instability.
        if sum.abs() <= f64::EPSILON {
            None
        } else {
            Some((q3_val - q1_val) / sum)
        }
    } else {
        None
    }
}

/// Compute Z-Score of Mode: (mode - mean) / stddev
#[inline]
fn compute_mode_zscore(mode: Option<f64>, mean: Option<f64>, stddev: Option<f64>) -> Option<f64> {
    if let (Some(mode_val), Some(mean_val), Some(stddev_val)) = (mode, mean, stddev) {
        if stddev_val.abs() > f64::EPSILON {
            Some((mode_val - mean_val) / stddev_val)
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute Relative Standard Error: sem / mean
#[inline]
fn compute_relative_standard_error(sem: Option<f64>, mean: Option<f64>) -> Option<f64> {
    if let (Some(sem_val), Some(mean_val)) = (sem, mean) {
        if mean_val.abs() > f64::EPSILON {
            Some(sem_val / mean_val)
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute Z-Score: (value - mean) / stddev
#[inline]
fn compute_zscore(value: Option<f64>, mean: Option<f64>, stddev: Option<f64>) -> Option<f64> {
    if let (Some(val), Some(mean_val), Some(stddev_val)) = (value, mean, stddev) {
        if stddev_val.abs() > f64::EPSILON {
            Some((val - mean_val) / stddev_val)
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute Median-to-Mean Ratio: median / mean
#[inline]
fn compute_median_mean_ratio(median: Option<f64>, mean: Option<f64>) -> Option<f64> {
    if let (Some(median_val), Some(mean_val)) = (median, mean) {
        if mean_val.abs() > f64::EPSILON {
            Some(median_val / mean_val)
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute IQR-to-Range Ratio: iqr / range
#[inline]
fn compute_iqr_range_ratio(iqr: Option<f64>, range: Option<f64>) -> Option<f64> {
    if let (Some(iqr_val), Some(range_val)) = (iqr, range) {
        if range_val.abs() > f64::EPSILON {
            Some(iqr_val / range_val)
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute MAD-to-StdDev Ratio: mad / stddev
#[inline]
fn compute_mad_stddev_ratio(mad: Option<f64>, stddev: Option<f64>) -> Option<f64> {
    if let (Some(mad_val), Some(stddev_val)) = (mad, stddev) {
        if stddev_val.abs() > f64::EPSILON {
            Some(mad_val / stddev_val)
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute Trimean: (Q1 + 2*median + Q3) / 4
/// Tukey's trimean - a robust estimator of central tendency that
/// combines the median with the midhinge.
#[inline]
fn compute_trimean(q1: Option<f64>, median: Option<f64>, q3: Option<f64>) -> Option<f64> {
    if let (Some(q1_val), Some(median_val), Some(q3_val)) = (q1, median, q3) {
        Some((2.0f64.mul_add(median_val, q1_val) + q3_val) / 4.0)
    } else {
        None
    }
}

/// Compute Midhinge: (Q1 + Q3) / 2
/// Midpoint of the middle 50% of data, a robust central tendency measure.
#[inline]
const fn compute_midhinge(q1: Option<f64>, q3: Option<f64>) -> Option<f64> {
    if let (Some(q1_val), Some(q3_val)) = (q1, q3) {
        Some(f64::midpoint(q1_val, q3_val))
    } else {
        None
    }
}

/// Compute Robust Coefficient of Variation: MAD / |median|
/// Uses robust measures (MAD and median magnitude) instead of stddev and mean.
#[inline]
fn compute_robust_cv(mad: Option<f64>, median: Option<f64>) -> Option<f64> {
    if let (Some(mad_val), Some(median_val)) = (mad, median) {
        if median_val.abs() > f64::EPSILON {
            Some(mad_val / median_val.abs())
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute Jarque-Bera test statistic: (n/6) * (S^2 + K^2/4)
/// Tests whether data follows a normal distribution.
/// Returns (jb_statistic, p_value) where p_value is from chi-squared(2) distribution.
#[inline]
fn compute_jarque_bera(skewness: Option<f64>, kurtosis: Option<f64>, n: u64) -> Option<(f64, f64)> {
    if n < 3 {
        return None;
    }
    if let (Some(skew_val), Some(kurt_val)) = (skewness, kurtosis) {
        #[allow(clippy::cast_precision_loss)]
        let n_f64 = n as f64;
        let jb = (n_f64 / 6.0) * skew_val.mul_add(skew_val, kurt_val * kurt_val / 4.0);
        // Upper-tail p-value from chi-squared distribution with 2 degrees of freedom
        // For chi-squared(2), the survival function (1 - CDF) is e^(-x/2)
        let p_value = (-jb / 2.0_f64).exp();
        Some((jb, p_value))
    } else {
        None
    }
}

/// Compute Bimodality Coefficient: (skewness² + 1) / (kurtosis + 3)
/// BC < 0.555 indicates unimodal, BC >= 0.555 indicates bimodal/multimodal
fn compute_bimodality_coefficient(skewness: Option<f64>, kurtosis: Option<f64>) -> Option<f64> {
    if let (Some(skew_val), Some(kurt_val)) = (skewness, kurtosis) {
        let denominator = kurt_val + 3.0;
        if denominator.abs() > f64::EPSILON {
            Some(skew_val.mul_add(skew_val, 1.0) / denominator)
        } else {
            None
        }
    } else {
        None
    }
}

/// Compute Normalized Entropy: shannon_entropy / log2(cardinality)
/// Values range from 0 (all values identical) to 1 (all values equally distributed)
fn compute_normalized_entropy(
    shannon_entropy: Option<f64>,
    cardinality: Option<u64>,
) -> Option<f64> {
    if let (Some(entropy_val), Some(card_val)) = (shannon_entropy, cardinality) {
        if card_val > 1 {
            #[allow(clippy::cast_precision_loss)]
            let max_entropy = (card_val as f64).log2();
            if max_entropy.abs() > f64::EPSILON {
                Some(entropy_val / max_entropy)
            } else {
                None
            }
        } else {
            // If cardinality is 0 or 1, normalized entropy is 0
            Some(0.0)
        }
    } else {
        None
    }
}

/// Parse a numeric value from a string, handling empty strings and invalid values
#[inline]
fn parse_float_opt(s: &str) -> Option<f64> {
    if s.is_empty() {
        return None;
    }
    // Filter NaN/±Infinity: a single non-finite cell would poison Welford
    // correlation state and propagate silently through all downstream statistics.
    fast_float2::parse::<f64, &[u8]>(s.as_bytes())
        .ok()
        .filter(|v| v.is_finite())
}

/// Parse a numeric value from bytes, handling empty bytes and invalid values
#[inline]
fn parse_float_opt_from_bytes(bytes: &[u8]) -> Option<f64> {
    if bytes.is_empty() {
        return None;
    }
    fast_float2::parse::<f64, &[u8]>(bytes)
        .ok()
        .filter(|v| v.is_finite())
}

/// Format a percentile value compactly: integral percentiles render without a
/// fractional part (e.g. `5.0` -> `"5"`, `5.5` -> `"5.5"`). Used for both
/// constructing column names and for looking up keys in the percentiles string;
/// keeping a single formatter avoids drift between the two sites.
#[inline]
fn fmt_pct(p: f64) -> String {
    #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
    if p.fract() == 0.0 && p >= 0.0 {
        format!("{}", p as u32)
    } else {
        format!("{p}")
    }
}

/// Parse a percentile value from the percentiles column string
/// Format: "5: value1|10: value2|..." (separator from QSV_STATS_SEPARATOR env var, default "|")
/// For Date/DateTime types, values are RFC3339 date strings; for numeric types, they're numbers
/// Returns the numeric value (in days since epoch for dates) for the specified percentile label, or
/// None if not found
fn parse_percentile_value(
    percentile_str: &str,
    percentile_label: &str,
    field_type: FieldType,
    separator: &str,
    prefer_dmy: bool,
) -> Option<f64> {
    if percentile_str.is_empty() {
        return None;
    }

    // Split by separator and find matching percentile
    for entry in percentile_str.split(separator) {
        let entry = entry.trim();
        if let Some(colon_pos) = entry.find(':') {
            let label = entry[..colon_pos].trim();
            let value_str = entry[colon_pos + 1..].trim();

            if label == percentile_label {
                // For Date/DateTime types, parse as date string; for numeric types, parse as float
                return if field_type.is_date_or_datetime() {
                    parse_date_to_days(value_str, prefer_dmy)
                } else {
                    parse_float_opt(value_str)
                };
            }
        }
    }

    None
}

/// Parse all percentile string values from the percentiles column string
/// Format: "5: value1|10: value2|25: value3|..." (separator from QSV_STATS_SEPARATOR env var,
/// default "|") Returns a vector of all percentile value strings (the values after colons)
/// Used for pattern matching all percentile values in fast mode
fn parse_all_percentile_string_values<'a>(
    percentile_str: &'a str,
    separator: &str,
) -> Vec<&'a str> {
    if percentile_str.is_empty() {
        return Vec::new();
    }

    // Split by separator and extract all values after colons
    percentile_str
        .split(separator)
        .filter_map(|entry| {
            let entry = entry.trim();
            if let Some(colon_pos) = entry.find(':') {
                let value_str = entry[colon_pos + 1..].trim();
                if !value_str.is_empty() {
                    return Some(value_str);
                }
            }
            None
        })
        .collect()
}

/// Field type enum for efficient comparisons
/// Matches the FieldType enum from stats.rs but kept local for performance
#[allow(clippy::enum_variant_names)]
#[derive(Clone, Copy, PartialEq)]
enum FieldType {
    TNull,
    TString,
    TFloat,
    TInteger,
    TDate,
    TDateTime,
    TBoolean,
}

impl FieldType {
    /// Convert string representation to FieldType enum
    /// Returns None if the string doesn't match any known type
    #[inline]
    fn from_str(s: &str) -> Option<FieldType> {
        match s {
            "NULL" => Some(FieldType::TNull),
            "String" => Some(FieldType::TString),
            "Float" => Some(FieldType::TFloat),
            "Integer" => Some(FieldType::TInteger),
            "Date" => Some(FieldType::TDate),
            "DateTime" => Some(FieldType::TDateTime),
            "Boolean" => Some(FieldType::TBoolean),
            _ => None,
        }
    }

    /// Check if this type is numeric or date/datetime
    #[inline]
    const fn is_numeric_or_date_type(self) -> bool {
        matches!(
            self,
            FieldType::TInteger
                | FieldType::TFloat
                | FieldType::TDate
                | FieldType::TDateTime
                | FieldType::TBoolean
        )
    }

    /// Check if this type is Date or DateTime
    #[inline]
    const fn is_date_or_datetime(self) -> bool {
        matches!(self, FieldType::TDate | FieldType::TDateTime)
    }
}

/// Parse a date/datetime value and convert to days since epoch
/// Returns None if parsing fails or value is empty
#[inline]
fn parse_date_to_days(s: &str, prefer_dmy: bool) -> Option<f64> {
    if s.is_empty() {
        return None;
    }
    #[allow(clippy::cast_precision_loss)]
    parse_with_preference(s, prefer_dmy)
        .ok()
        .map(|dt| dt.timestamp_millis() as f64 / 86_400_000.0)
}

/// Detect Gregorian date types (gYearMonth, gYear, gMonthDay, gDay, gMonth) using
/// optimized pattern matching with cheap checks first, regex only when necessary.
/// Returns Some("typeName?") or Some("typeName??") if detected, None otherwise.
/// Performance optimized: uses numeric comparisons for Integer gYear, cheap string
/// checks (length/prefix) before regex for String types.
/// Quick mode checks min/max values only (fast), thorough mode checks all percentile values (slower
/// but more confident).
fn detect_gregorian_date_type(
    min_str: Option<&str>,
    max_str: Option<&str>,
    field_type_str: &str,
    min_val: Option<f64>,
    max_val: Option<f64>,
    scan_mode: &str,
    percentile_values: Option<&[&str]>,
) -> Option<String> {
    // Determine suffix based on scan mode
    // More question marks = less confidence
    let suffix = if scan_mode == "quick" { "??" } else { "?" };

    // Shared closure used for both quick and thorough modes
    // to check if a string matches a Gregorian date pattern
    let check_value = |s: &str| -> Option<&str> {
        // gYearMonth: "1999-05" (length 7, dash at position 4)
        if s.len() == 7
            && s.as_bytes().get(4) == Some(&b'-')
            && regex_oncelock!(r"^\d{4}-(0[1-9]|1[0-2])$").is_match(s)
        {
            // Validate that the month portion is within 01-12
            let month_str = &s[5..7];
            if let Ok(month) = month_str.parse::<u8>()
                && (1..=12).contains(&month)
            {
                return Some("gYearMonth");
            }
        }

        // gYear: "1999" (length 4)
        if s.len() == 4
            && regex_oncelock!(r"^\d{4}$").is_match(s)
            && let Ok(year) = s.parse::<i32>()
            && (1000..=3000).contains(&year)
        {
            return Some("gYear");
        }

        // gMonthDay: "--05-01" (length 7)
        if s.len() == 7 && regex_oncelock!(r"^--\d{2}-\d{2}$").is_match(s) {
            // validate numeric ranges: month 1-12, with month-specific day limits
            if let (Ok(month), Ok(day)) = (s[2..4].parse::<u32>(), s[5..7].parse::<u32>())
                && (1..=12).contains(&month)
                && match month {
                    // Months with 31 days
                    1 | 3 | 5 | 7 | 8 | 10 | 12 => (1..=31).contains(&day),
                    // Months with 30 days
                    4 | 6 | 9 | 11 => (1..=30).contains(&day),
                    // February: allow up to 29 to accommodate leap years (year is unknown)
                    2 => (1..=29).contains(&day),
                    _ => false,
                }
            {
                return Some("gMonthDay");
            }
        }

        // gDay: "---01" (length 5)
        if s.len() == 5 && regex_oncelock!(r"^---\d{2}$").is_match(s) &&
            // validate numeric range: day 1-31
            let Ok(day) = s[3..5].parse::<u32>()
            && (1..=31).contains(&day)
        {
            return Some("gDay");
        }

        // gMonth: "--05" (length 4)
        if s.len() == 4 && regex_oncelock!(r"^--\d{2}$").is_match(s) {
            // validate numeric range: month 1-12
            if let Ok(month) = s[2..4].parse::<u32>()
                && (1..=12).contains(&month)
            {
                return Some("gMonth");
            }
        }

        None
    };

    // Thorough mode: check all percentile values
    if scan_mode == "thorough" {
        if let Some(pct_values) = percentile_values {
            if pct_values.is_empty() {
                return None;
            }

            // Fast path for Integer gYear (no regex needed)
            if field_type_str == "Integer" {
                // Parse all percentile values as numbers and check if all are in year range
                // Skip empty strings but require all non-empty values to be in range
                let non_empty_values: Vec<&str> = pct_values
                    .iter()
                    .copied()
                    .filter(|&s| !s.is_empty())
                    .collect();
                if !non_empty_values.is_empty() {
                    let all_in_range = non_empty_values.iter().all(|&val_str| {
                        if let Some(val) = parse_float_opt(val_str) {
                            (1000.0..=3000.0).contains(&val)
                        } else {
                            false
                        }
                    });
                    if all_in_range {
                        return Some(format!("gYear{suffix}"));
                    }
                }
                return None;
            }

            // For String types, check all percentile values against patterns
            // Check all percentile values - only return type if ALL match the same pattern
            let mut matched_type: Option<&str> = None;
            for &val_str in pct_values {
                if val_str.is_empty() {
                    continue; // Skip empty values
                }
                {
                    let pattern_type = check_value(val_str)?;
                    match matched_type {
                        None => matched_type = Some(pattern_type),
                        Some(existing_type) if existing_type == pattern_type => {
                            // Same pattern, continue
                        },
                        _ => {
                            // Different pattern or no match, not consistent
                            return None;
                        },
                    }
                }
            }

            // All values matched the same pattern
            if let Some(base_type) = matched_type {
                return Some(format!("{base_type}{suffix}"));
            }
        }
        return None;
    }

    // Quick mode: check min/max values
    // Fast path for Integer gYear (no regex needed)
    if field_type_str == "Integer" {
        if let (Some(min), Some(max)) = (min_val, max_val) {
            // Check if values are in reasonable year range (1000-3000)
            if min >= 1000.0 && max <= 3000.0 {
                return Some(format!("gYear{suffix}"));
            }
        }
        // Not a year range, return None to continue with normal Integer inference
        return None;
    }

    // For String types, check both min and max to increase confidence
    // Check min_str first
    if let Some(min_s) = min_str
        && !min_s.is_empty()
        && let Some(greg_type) = check_value(min_s)
    {
        // If max_str is available, verify it matches the same pattern for confidence
        if let Some(max_s) = max_str {
            if !max_s.is_empty() {
                if let Some(max_type) = check_value(max_s) {
                    // Both match the same type, return it
                    if greg_type == max_type {
                        return Some(format!("{greg_type}{suffix}"));
                    }
                    // Different patterns, not confident - return None
                    return None;
                }
                // max_str does not match pattern, don't return based only on min_str
                return None;
            }
            // max_str is empty; treat as missing, don't return based only on min_str
            return None;
        }
        // max_str not present at all, rely on min_str alone (conservative)
        return Some(format!("{greg_type}{suffix}"));
    }

    // Check max_str if min_str didn't match
    if let Some(max_s) = max_str
        && !max_s.is_empty()
        && let Some(greg_type) = check_value(max_s)
    {
        return Some(format!("{greg_type}{suffix}"));
    }

    None
}

/// Infer the most specific W3C XML Schema datatype based on field type and min/max values
/// Returns the XSD type string (e.g., "byte", "int", "decimal", "string", "date", etc.)
/// Based on the analysis at https://github.com/user-attachments/files/23841656/xsd_analysis.md
fn infer_xsd_type(
    field_type_str: &str,
    min_val: Option<f64>,
    max_val: Option<f64>,
    field_type_enum: Option<FieldType>,
    scan_mode: &str,
    min_str: Option<&str>,
    max_str: Option<&str>,
    percentile_values: Option<&[&str]>,
) -> String {
    // Handle NULL type
    if field_type_str == "NULL" || field_type_str.is_empty() {
        return String::new();
    }

    // Handle Boolean type
    if field_type_str == "Boolean" {
        return "boolean".to_string();
    }

    // Check for Gregorian date types early (after NULL/Boolean, before other type checks)
    // This allows detection for both Integer and String fields
    if let Some(greg_type) = detect_gregorian_date_type(
        min_str,
        max_str,
        field_type_str,
        min_val,
        max_val,
        scan_mode,
        percentile_values,
    ) {
        return greg_type;
    }

    // Handle Date and DateTime types
    if field_type_enum == Some(FieldType::TDate) {
        return "date".to_string();
    }
    if field_type_enum == Some(FieldType::TDateTime) {
        return "dateTime".to_string();
    }

    // Handle String type
    if field_type_str == "String" {
        return "string".to_string();
    }

    // Handle Float type
    if field_type_str == "Float" {
        return "decimal".to_string();
    }

    // Handle Integer type with range-based refinement
    if field_type_str == "Integer" {
        let (Some(min), Some(max)) = (min_val, max_val) else {
            // If min/max not available, default to integer
            return "integer".to_string();
        };

        // Check for unsigned integer types first (most specific first)
        // Only check unsigned types if min >= 0
        if min >= 0.0 {
            if max <= 255.0 {
                return "unsignedByte".to_string();
            }
            if max <= 65_535.0 {
                return "unsignedShort".to_string();
            }
            if max <= 4_294_967_295.0 {
                return "unsignedInt".to_string();
            }
            // unsignedLong: 0 to 2^64-1 (18446744073709551615)
            // Check if max fits in u64 range
            if max <= 18_446_744_073_709_551_615.0 {
                return "unsignedLong".to_string();
            }
            // Check for special unsigned constraints (unbounded)
            if min > 0.0 {
                return "positiveInteger".to_string();
            }
            // min >= 0.0 (already checked above)
            return "nonNegativeInteger".to_string();
        }

        // Check for signed integer types (most specific first)
        // Only check signed types if min < 0 (or if we have negative values)
        // Use f64 comparisons to avoid clamping issues
        if min >= -128.0 && max <= 127.0 {
            return "byte".to_string();
        }
        if min >= -32_768.0 && max <= 32_767.0 {
            return "short".to_string();
        }
        if min >= -2_147_483_648.0 && max <= 2_147_483_647.0 {
            return "int".to_string();
        }
        if min >= -9_223_372_036_854_775_808.0 && max <= 9_223_372_036_854_775_807.0 {
            return "long".to_string();
        }

        // Check for special signed integer constraints
        if max < 0.0 {
            return "negativeInteger".to_string();
        }
        if max <= 0.0 {
            return "nonPositiveInteger".to_string();
        }

        // Default to unbounded integer
        return "integer".to_string();
    }

    // Fallback: return empty string for unrecognized types
    String::new()
}

/// Convert days since epoch to RFC3339 formatted date string
/// For Date types, returns only the date component (YYYY-MM-DD)
/// For DateTime types, returns full RFC3339 format with time and timezone
fn days_to_rfc3339(days: f64, field_type: FieldType) -> String {
    // Convert days to milliseconds
    #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
    let timestamp_ms = (days * 86_400_000.0) as i64;

    let date_val = chrono::DateTime::from_timestamp_millis(timestamp_ms)
        .unwrap_or_default()
        .to_rfc3339();

    // if type = Date, only return the date component
    if field_type == FieldType::TDate {
        return date_val[..10].to_string();
    }
    date_val
}

/// Field information needed for outlier counting and winsorized/trimmed means
#[derive(Clone)]
struct OutlierFieldInfo {
    col_idx:         usize,
    field_type:      FieldType, // Use enum for faster comparisons
    lower_outer:     f64,
    lower_inner:     f64,
    upper_inner:     f64,
    upper_outer:     f64,
    lower_threshold: f64, // For winsorization/trimming (Q1 or percentile)
    upper_threshold: f64, // For winsorization/trimming (Q3 or percentile)
}

// Indices into `OutlierStats::counts` (keep in sync with OUTLIER_COUNTS_LEN).
const OUTLIER_EXTREME_LOWER: usize = 0;
const OUTLIER_MILD_LOWER: usize = 1;
const OUTLIER_NORMAL: usize = 2;
const OUTLIER_MILD_UPPER: usize = 3;
const OUTLIER_EXTREME_UPPER: usize = 4;
const OUTLIER_TOTAL: usize = 5;
const OUTLIER_COUNTS_LEN: usize = 6;

/// Statistics tracked during outlier scanning
#[derive(Clone, Default)]
struct OutlierStats {
    // Counts indexed by OUTLIER_{EXTREME_LOWER, MILD_LOWER, NORMAL, MILD_UPPER, EXTREME_UPPER,
    // TOTAL}
    counts:                 [u64; OUTLIER_COUNTS_LEN],
    // Sums
    sum_outliers:           f64,
    sum_normal:             f64,
    sum_all:                f64,
    // Min/Max
    min_outliers:           Option<f64>,
    max_outliers:           Option<f64>,
    min_normal:             Option<f64>,
    max_normal:             Option<f64>,
    // Winsorized and trimmed means
    winsorized_sum:         f64,
    winsorized_count:       u64,
    trimmed_sum:            f64,
    trimmed_count:          u64,
    // For variance/stddev computation (using sum of squares)
    sum_squares_outliers:   f64,
    sum_squares_normal:     f64,
    sum_squares_trimmed:    f64,
    sum_squares_winsorized: f64,
    // For trimmed/winsorized range
    min_trimmed:            Option<f64>,
    max_trimmed:            Option<f64>,
    min_winsorized:         Option<f64>,
    max_winsorized:         Option<f64>,
    // Total count of all values processed
    count_all:              u64,
}

/// Statistics for Kurtosis, Gini & Atkinson Index
#[derive(Clone, Default)]
struct KGAStats {
    kurtosis:         Option<f64>,
    gini_coefficient: Option<f64>,
    atkinson_index:   Option<f64>,
    theil_index:      Option<f64>,
    mean_ad:          Option<f64>,
}

/// Statistics for Shannon Entropy and Simpson's Diversity Index
#[derive(Clone, Default)]
struct EntropyStats {
    entropy:            Option<f64>,
    simpsons_diversity: Option<f64>,
}

/// Online algorithm state for correlation/covariance computation
/// Uses Welford's online algorithm for aggregating across chunks
#[derive(Clone, Default)]
struct CorrelationState {
    count:  u64,
    mean_x: f64,
    mean_y: f64,
    m2_x:   f64, // sum of squared differences for x
    m2_y:   f64, // sum of squared differences for y
    cxy:    f64, // sum of (x - mean_x) * (y - mean_y)
}

/// Statistics tracked during bivariate computation for a field pair
#[derive(Clone, Default)]
struct BivariateChunkStats {
    correlation_state: CorrelationState,
    x_values:          Vec<f64>, // For Spearman/Kendall (need ranks)
    y_values:          Vec<f64>, // For Spearman/Kendall (need ranks)
    // Frequency counts for mutual information (more memory efficient than storing all strings)
    xy_counts:         HashMap<(String, String), u64>, // Joint frequencies
    x_counts:          HashMap<String, u64>,           // Marginal frequencies for x
    y_counts:          HashMap<String, u64>,           // Marginal frequencies for y
    total_pairs:       u64,                            // Total count of pairs
}

/// Final bivariate statistics for a field pair
#[derive(Clone, Default)]
struct BivariateStats {
    pearson: Option<f64>,
    spearman: Option<f64>,
    kendall: Option<f64>,
    covariance_sample: Option<f64>,
    covariance_population: Option<f64>,
    mutual_information: Option<f64>,
    normalized_mutual_information: Option<f64>,
    n_pairs: u64,
}

/// Field information for bivariate computation
#[derive(Clone)]
struct BivariateFieldInfo {
    col_idx:     usize,
    field_type:  FieldType,
    // Pre-computed statistics from stats CSV (used for optimizations)
    stddev:      Option<f64>, // Pre-computed standard deviation (used for filtering)
    variance:    Option<f64>, // Pre-computed variance (used for filtering)
    cardinality: Option<u64>, // Pre-computed cardinality (used for threshold filtering)
}

/// Update correlation state with a new pair of values using Welford's online algorithm
#[inline]
#[allow(clippy::cast_precision_loss)]
fn update_correlation_state(state: &mut CorrelationState, x: f64, y: f64) {
    state.count += 1;
    let n = state.count as f64;

    let delta_x = x - state.mean_x;
    let delta_y = y - state.mean_y;

    // Update means
    state.mean_x += delta_x / n;
    state.mean_y += delta_y / n;

    // Update sum of squared differences and covariance term
    let delta_x_new = x - state.mean_x;
    let delta_y_new = y - state.mean_y;

    state.m2_x = delta_x.mul_add(delta_x_new, state.m2_x);
    state.m2_y = delta_y.mul_add(delta_y_new, state.m2_y);
    state.cxy = delta_x.mul_add(delta_y_new, state.cxy);
}

/// Merge two correlation states (for aggregating across chunks)
#[allow(clippy::cast_precision_loss)]
fn merge_correlation_states(
    state1: &CorrelationState,
    state2: &CorrelationState,
) -> CorrelationState {
    if state1.count == 0 {
        return state2.clone();
    }
    if state2.count == 0 {
        return state1.clone();
    }

    let n1 = state1.count as f64;
    let n2 = state2.count as f64;
    let n_total = n1 + n2;

    // NOTE: we use fused multiply-add extensively below
    // for more efficient, performant, accurate computations.
    // the original formula is in a comment above each FMA implementation.

    // Combined mean
    // let mean_x_combined = (state1.mean_x * n1 + state2.mean_x * n2) / n_total;
    let mean_x_combined = state1.mean_x.mul_add(n1, state2.mean_x * n2) / n_total;
    // let mean_y_combined = (state1.mean_y * n1 + state2.mean_y * n2) / n_total;
    let mean_y_combined = state1.mean_y.mul_add(n1, state2.mean_y * n2) / n_total;

    // Combined variance terms (using parallel algorithm formula)
    let delta_x1 = state1.mean_x - mean_x_combined;
    let delta_x2 = state2.mean_x - mean_x_combined;
    let delta_y1 = state1.mean_y - mean_y_combined;
    let delta_y2 = state2.mean_y - mean_y_combined;

    let m2_x_combined =
        // state1.m2_x + state2.m2_x + delta_x1 * delta_x1 * n1 + delta_x2 * delta_x2 * n2;
        (delta_x2 * delta_x2).mul_add(n2, (delta_x1 * delta_x1).mul_add(n1, state1.m2_x + state2.m2_x));
    let m2_y_combined =
        // state1.m2_y + state2.m2_y + delta_y1 * delta_y1 * n1 + delta_y2 * delta_y2 * n2;
        (delta_y2 * delta_y2).mul_add(n2, (delta_y1 * delta_y1).mul_add(n1, state1.m2_y + state2.m2_y));

    // Combined covariance term
    let cxy_combined =
        // state1.cxy + state2.cxy + delta_x1 * delta_y1 * n1 + delta_x2 * delta_y2 * n2;
        (delta_x2 * delta_y2).mul_add(n2, (delta_x1 * delta_y1).mul_add(n1, state1.cxy + state2.cxy));

    CorrelationState {
        count:  state1.count + state2.count,
        mean_x: mean_x_combined,
        mean_y: mean_y_combined,
        m2_x:   m2_x_combined,
        m2_y:   m2_y_combined,
        cxy:    cxy_combined,
    }
}

/// Compute final Pearson correlation coefficient from correlation state
#[allow(clippy::cast_precision_loss)]
fn finalize_pearson_correlation(state: &CorrelationState) -> Option<f64> {
    if state.count < 2 {
        return None;
    }

    let variance_x = state.m2_x / (state.count as f64 - 1.0);
    let variance_y = state.m2_y / (state.count as f64 - 1.0);

    if variance_x <= 0.0 || variance_y <= 0.0 {
        return None;
    }

    let covariance = state.cxy / (state.count as f64 - 1.0);
    let stddev_x = variance_x.sqrt();
    let stddev_y = variance_y.sqrt();

    if stddev_x.abs() > f64::EPSILON && stddev_y.abs() > f64::EPSILON {
        Some(covariance / (stddev_x * stddev_y))
    } else {
        None
    }
}

/// Compute final covariance from correlation state
#[allow(clippy::cast_precision_loss)]
fn finalize_covariance(state: &CorrelationState, sample: bool) -> Option<f64> {
    if state.count < 2 {
        return None;
    }

    let divisor = if sample {
        state.count as f64 - 1.0
    } else {
        state.count as f64
    };

    Some(state.cxy / divisor)
}

/// Compute Pearson correlation coefficient from two arrays of values
fn compute_pearson_correlation(x: &[f64], y: &[f64]) -> Option<f64> {
    if x.len() != y.len() || x.len() < 2 {
        return None;
    }

    let mut state = CorrelationState::default();
    for (xi, yi) in x.iter().zip(y.iter()) {
        update_correlation_state(&mut state, *xi, *yi);
    }

    finalize_pearson_correlation(&state)
}

/// Compute Spearman's rank correlation coefficient
#[allow(clippy::cast_precision_loss)]
#[allow(clippy::many_single_char_names)]
fn compute_spearman_correlation(x: &[f64], y: &[f64]) -> Option<f64> {
    if x.len() != y.len() || x.len() < 2 {
        return None;
    }

    let n = x.len();

    // Pre-allocate with capacity to avoid reallocations
    let mut x_ranked: Vec<(usize, f64)> = Vec::with_capacity(n);
    x_ranked.extend(x.iter().enumerate().map(|(i, &v)| (i, v)));

    let mut y_ranked: Vec<(usize, f64)> = Vec::with_capacity(n);
    y_ranked.extend(y.iter().enumerate().map(|(i, &v)| (i, v)));

    // Use total_cmp for faster, more predictable sorting (handles NaNs consistently)
    // This is faster than partial_cmp and gives consistent ordering
    x_ranked.sort_unstable_by(|a, b| a.1.total_cmp(&b.1));
    y_ranked.sort_unstable_by(|a, b| a.1.total_cmp(&b.1));

    // Pre-allocate rank vectors
    let mut x_ranks = vec![0.0; n];
    let mut y_ranks = vec![0.0; n];

    // Rank x values (handle ties by averaging) - optimized loop
    let mut i = 0;
    while i < n {
        let mut j = i;
        let val = x_ranked[i].1;
        // Use total_cmp for tie detection - faster than abs diff
        while j < n && x_ranked[j].1.total_cmp(&val) == std::cmp::Ordering::Equal {
            j += 1;
        }
        let rank = (i + j - 1) as f64 / 2.0 + 1.0;
        // Use slice assignment for better cache locality
        for k in i..j {
            x_ranks[x_ranked[k].0] = rank;
        }
        i = j;
    }

    // Rank y values - use total_cmp for faster comparison
    i = 0;
    while i < n {
        let mut j = i;
        let val = y_ranked[i].1;
        while j < n && y_ranked[j].1.total_cmp(&val) == std::cmp::Ordering::Equal {
            j += 1;
        }
        let rank = (i + j - 1) as f64 / 2.0 + 1.0;
        for k in i..j {
            y_ranks[y_ranked[k].0] = rank;
        }
        i = j;
    }

    // Compute Pearson correlation on ranks
    compute_pearson_correlation(&x_ranks, &y_ranks)
}

/// Count inversions in y values when sorted by x using merge sort (O(n log n))
/// Returns number of inversions (discordant pairs)
#[allow(clippy::cast_precision_loss)]
fn count_inversions_merge(
    pairs: &mut [(f64, f64)],
    temp: &mut [(f64, f64)],
    left: usize,
    right: usize,
) -> i64 {
    if left >= right {
        return 0;
    }

    let mid = left + (right - left) / 2;
    let mut inversions = count_inversions_merge(pairs, temp, left, mid)
        + count_inversions_merge(pairs, temp, mid + 1, right);

    // Merge and count inversions - use total_cmp for faster comparison
    let mut i = left;
    let mut j = mid + 1;
    let mut k = left;

    while i <= mid && j <= right {
        // Use total_cmp instead of <= for faster comparison
        if pairs[i].1.total_cmp(&pairs[j].1) == std::cmp::Ordering::Greater {
            // Inversion found: pairs[i].1 > pairs[j].1
            // All remaining elements in left half form inversions with pairs[j]
            inversions += (mid - i + 1) as i64;
            temp[k] = pairs[j];
            j += 1;
        } else {
            // No inversion: pairs[i].1 <= pairs[j].1
            // Copy pairs[i] to temp and move to next element in left half
            temp[k] = pairs[i];
            i += 1;
        }
        k += 1; // Move to next position in temp array
    }

    // Copy remaining elements - use copy_from_slice for better performance
    if i <= mid {
        let remaining = mid - i + 1;
        temp[k..k + remaining].copy_from_slice(&pairs[i..i + remaining]);
    }
    if j <= right {
        let remaining = right - j + 1;
        temp[k..k + remaining].copy_from_slice(&pairs[j..j + remaining]);
    }

    // Copy back from temp
    pairs[left..=right].copy_from_slice(&temp[left..=right]);

    inversions
}

/// Compute Kendall's tau rank correlation coefficient using O(n log n) merge sort
#[allow(clippy::cast_precision_loss)]
#[allow(clippy::many_single_char_names)]
fn compute_kendall_tau(x: &[f64], y: &[f64]) -> Option<f64> {
    if x.len() != y.len() || x.len() < 2 {
        return None;
    }

    let n = x.len() as f64;
    let pairs_len = x.len();

    // Pre-allocate indices vector
    let mut y_indices: Vec<usize> = Vec::with_capacity(pairs_len);
    y_indices.extend(0..pairs_len);

    // Use total_cmp for faster, more predictable sorting
    y_indices.sort_unstable_by(|&a, &b| y[a].total_cmp(&y[b]).then_with(|| x[a].total_cmp(&x[b])));

    // Count ties in y
    let mut ties_y = 0i64;
    let mut i = 0;
    while i < pairs_len {
        let mut j = i + 1;
        let val = y[y_indices[i]];
        // Use total_cmp instead of abs diff for tie detection
        while j < pairs_len && y[y_indices[j]].total_cmp(&val) == std::cmp::Ordering::Equal {
            j += 1;
        }
        let tie_count = (j - i) as i64;
        if tie_count > 1 {
            ties_y += tie_count * (tie_count - 1) / 2;
        }
        i = j;
    }

    // Pre-allocate pairs vector with capacity
    let mut pairs: Vec<(f64, f64)> = Vec::with_capacity(pairs_len);
    pairs.extend(x.iter().zip(y.iter()).map(|(&a, &b)| (a, b)));

    // Use total_cmp for faster sorting
    pairs.sort_unstable_by(|a, b| a.0.total_cmp(&b.0).then_with(|| a.1.total_cmp(&b.1)));

    // Count ties in x
    let mut ties_x = 0i64;
    i = 0;
    while i < pairs_len {
        let mut j = i + 1;
        let val = pairs[i].0;
        while j < pairs_len && pairs[j].0.total_cmp(&val) == std::cmp::Ordering::Equal {
            j += 1;
        }
        let tie_count = (j - i) as i64;
        if tie_count > 1 {
            ties_x += tie_count * (tie_count - 1) / 2;
        }
        i = j;
    }

    // Pre-allocate temp buffer once
    let mut temp = vec![(0.0, 0.0); pairs_len];
    let inversions = count_inversions_merge(&mut pairs, &mut temp, 0, pairs_len - 1);

    // Calculate concordant and discordant pairs
    let total_pairs = (n * (n - 1.0) / 2.0) as i64;
    let discordant = inversions;
    let concordant = total_pairs - discordant - ties_x - ties_y;

    let n0 = n * (n - 1.0) / 2.0;
    let n1 = ties_x as f64;
    let n2 = ties_y as f64;

    // Clamp factors to >= 0 to guard against tiny negative values from rounding
    // on tie-heavy data, which would otherwise produce sqrt(NaN).
    let denominator = ((n0 - n1).max(0.0) * (n0 - n2).max(0.0)).sqrt();

    if denominator.abs() < f64::EPSILON {
        return None;
    }

    let tau = ((concordant - discordant) as f64) / denominator;
    Some(tau)
}

/// Compute mutual information between two categorical/numeric fields from frequency counts
#[allow(clippy::cast_precision_loss)]
fn compute_mutual_information_from_counts(
    xy_counts: &HashMap<(String, String), u64>,
    x_counts: &HashMap<String, u64>,
    y_counts: &HashMap<String, u64>,
    total: u64,
) -> Option<f64> {
    if total == 0 {
        return None;
    }

    let total_f64 = total as f64;

    // Compute mutual information: MI(X,Y) = sum(p(x,y) * log2(p(x,y) / (p(x) * p(y))))
    let mut mi = 0.0;
    for ((x_val, y_val), &xy_count) in xy_counts {
        let p_xy = xy_count as f64 / total_f64;
        let p_x = x_counts.get(x_val).copied().unwrap_or(0) as f64 / total_f64;
        let p_y = y_counts.get(y_val).copied().unwrap_or(0) as f64 / total_f64;

        if p_x > 0.0 && p_y > 0.0 && p_xy > 0.0 {
            mi = p_xy.mul_add((p_xy / (p_x * p_y)).log2(), mi);
        }
    }

    Some(mi)
}

/// Compute Shannon entropy from frequency counts
/// Uses the same formula as compute_all_entropy(): H(X) = -Σ p_i * log2(p_i)
/// where p_i = count_i / total
#[allow(clippy::cast_precision_loss)]
fn compute_entropy_from_counts(counts: &HashMap<String, u64>, total: u64) -> Option<f64> {
    if total == 0 {
        return None;
    }

    let total_f64 = total as f64;
    let mut entropy = 0.0;

    for count in counts.values() {
        if *count > 0 {
            let p = *count as f64 / total_f64;
            entropy = p.mul_add(-p.log2(), entropy);
        }
    }

    Some(entropy)
}

/// Compute normalized mutual information from mutual information and entropies
/// NMI = MI / sqrt(H(X) * H(Y))
/// Returns None if either entropy is invalid (0, negative, or None) or if the denominator
/// is smaller than f64::EPSILON (guards against subnormal products from extremely low
/// entropies producing Inf/NaN).
fn compute_normalized_mutual_information(
    mi: Option<f64>,
    h_x: Option<f64>,
    h_y: Option<f64>,
) -> Option<f64> {
    let (Some(mi_val), Some(h_x_val), Some(h_y_val)) = (mi, h_x, h_y) else {
        return None;
    };

    // Check for invalid entropy values (non-positive)
    if h_x_val <= 0.0 || h_y_val <= 0.0 {
        return None;
    }

    // Compute denominator: sqrt(H(X) * H(Y))
    let denominator = (h_x_val * h_y_val).sqrt();
    if denominator < f64::EPSILON {
        return None;
    }

    // NMI = MI / sqrt(H(X) * H(Y))
    Some(mi_val / denominator)
}

/// Field information needed for Kurtosis, Gini & Atkinson Index computation (with precalculated
/// stats)
#[derive(Clone)]
struct KGAFieldInfo {
    col_idx:    usize,
    field_type: FieldType,
    mean:       Option<f64>,
    variance:   Option<f64>, // variance = stddev^2
    sum:        Option<f64>, // sum for Gini coefficient
}

/// Count outliers for a chunk of records and compute statistics
/// Returns a HashMap mapping field names to their outlier statistics
fn count_chunk_outliers<I>(
    fields_to_count: &HashMap<String, OutlierFieldInfo>,
    records: I,
) -> CliResult<HashMap<String, OutlierStats>>
where
    I: Iterator<Item = csv::Result<csv::ByteRecord>>,
{
    if fields_to_count.is_empty() {
        return Ok(HashMap::new());
    }

    // Initialize statistics for all fields
    let mut chunk_stats: HashMap<String, OutlierStats> = fields_to_count
        .keys()
        .map(|k| (k.clone(), OutlierStats::default()))
        .collect();

    let prefer_dmy = util::get_envvar_flag("QSV_PREFER_DMY");
    #[allow(unused_assignments)]
    let mut record: csv::ByteRecord = csv::ByteRecord::new();
    let mut value_bytes;
    let mut numeric_value;

    // Process each record in the chunk
    for result in records {
        record = result?;

        for (field_name, field_info) in fields_to_count {
            value_bytes = record.get(field_info.col_idx).unwrap_or(&[]);

            if value_bytes.is_empty() {
                continue; // Skip null/empty values
            }

            // Parse the value based on field type
            numeric_value = if field_info.field_type.is_date_or_datetime() {
                // Convert bytes to string for date parsing
                if let Ok(value_str) = from_utf8(value_bytes) {
                    parse_date_to_days(value_str, prefer_dmy)
                } else {
                    None
                }
            } else {
                parse_float_opt_from_bytes(value_bytes)
            };

            let Some(val) = numeric_value else {
                continue; // Skip values that can't be parsed
            };

            // Get mutable reference to stats for this field
            // safety: chunk_stats is pre-populated with all field names
            let Some(stats) = chunk_stats.get_mut(field_name) else {
                debug_assert!(false, "chunk_stats missing expected key: {field_name}");
                continue;
            };

            // Update sums and count
            stats.sum_all += val;
            stats.count_all += 1;

            // Compute winsorized and trimmed statistics
            let winsorized_val = val
                .max(field_info.lower_threshold)
                .min(field_info.upper_threshold);
            stats.winsorized_sum += winsorized_val;
            stats.winsorized_count += 1;
            // Track winsorized min/max and sum of squares
            stats.min_winsorized = Some(
                stats
                    .min_winsorized
                    .map_or(winsorized_val, |m| m.min(winsorized_val)),
            );
            stats.max_winsorized = Some(
                stats
                    .max_winsorized
                    .map_or(winsorized_val, |m| m.max(winsorized_val)),
            );
            stats.sum_squares_winsorized =
                winsorized_val.mul_add(winsorized_val, stats.sum_squares_winsorized);

            // For trimmed mean, only include values within thresholds
            if val >= field_info.lower_threshold && val <= field_info.upper_threshold {
                stats.trimmed_sum += val;
                stats.trimmed_count += 1;
                // Track trimmed min/max and sum of squares
                stats.min_trimmed = Some(stats.min_trimmed.map_or(val, |m| m.min(val)));
                stats.max_trimmed = Some(stats.max_trimmed.map_or(val, |m| m.max(val)));
                stats.sum_squares_trimmed = val.mul_add(val, stats.sum_squares_trimmed);
            }

            // Count outliers and track statistics based on fence comparisons
            if val < field_info.lower_outer {
                stats.counts[OUTLIER_EXTREME_LOWER] += 1;
                stats.counts[OUTLIER_TOTAL] += 1;
                stats.sum_outliers += val;
                stats.sum_squares_outliers = val.mul_add(val, stats.sum_squares_outliers);
                stats.min_outliers = Some(stats.min_outliers.map_or(val, |m| m.min(val)));
                stats.max_outliers = Some(stats.max_outliers.map_or(val, |m| m.max(val)));
            } else if val < field_info.lower_inner {
                stats.counts[OUTLIER_MILD_LOWER] += 1;
                stats.counts[OUTLIER_TOTAL] += 1;
                stats.sum_outliers += val;
                stats.sum_squares_outliers = val.mul_add(val, stats.sum_squares_outliers);
                stats.min_outliers = Some(stats.min_outliers.map_or(val, |m| m.min(val)));
                stats.max_outliers = Some(stats.max_outliers.map_or(val, |m| m.max(val)));
            } else if val <= field_info.upper_inner {
                stats.counts[OUTLIER_NORMAL] += 1;
                stats.sum_normal += val;
                stats.sum_squares_normal = val.mul_add(val, stats.sum_squares_normal);
                stats.min_normal = Some(stats.min_normal.map_or(val, |m| m.min(val)));
                stats.max_normal = Some(stats.max_normal.map_or(val, |m| m.max(val)));
            } else if val <= field_info.upper_outer {
                stats.counts[OUTLIER_MILD_UPPER] += 1;
                stats.counts[OUTLIER_TOTAL] += 1;
                stats.sum_outliers += val;
                stats.sum_squares_outliers = val.mul_add(val, stats.sum_squares_outliers);
                stats.min_outliers = Some(stats.min_outliers.map_or(val, |m| m.min(val)));
                stats.max_outliers = Some(stats.max_outliers.map_or(val, |m| m.max(val)));
            } else {
                stats.counts[OUTLIER_EXTREME_UPPER] += 1;
                stats.counts[OUTLIER_TOTAL] += 1;
                stats.sum_outliers += val;
                stats.sum_squares_outliers = val.mul_add(val, stats.sum_squares_outliers);
                stats.min_outliers = Some(stats.min_outliers.map_or(val, |m| m.min(val)));
                stats.max_outliers = Some(stats.max_outliers.map_or(val, |m| m.max(val)));
            }
        }
    }

    Ok(chunk_stats)
}

/// Count outliers for all fields, using parallel processing if index is available
/// Returns a HashMap mapping field names to their outlier statistics
fn count_all_outliers(
    fields_to_count: HashMap<String, OutlierFieldInfo>,
    input_path: &Path,
    flag_jobs: Option<usize>,
) -> CliResult<HashMap<String, OutlierStats>> {
    if fields_to_count.is_empty() {
        return Ok(HashMap::new());
    }

    // Check if index exists for parallel processing
    let input_path_str = input_path
        .to_str()
        .ok_or_else(|| CliError::Other(format!("Invalid input path: {}", input_path.display())))?;
    let input_path_string = input_path_str.to_string();
    let rconfig = Config::new(Some(&input_path_string));
    let indexed_result = rconfig.indexed()?;

    if let Some(idx) = indexed_result {
        // Parallel processing path
        let idx_count = idx.count() as usize;
        if idx_count == 0 {
            return Ok(HashMap::new());
        }

        // Only parallelize if file is large enough
        if idx_count < PARALLEL_THRESHOLD {
            // Fall back to sequential for small files
            let mut rdr = rconfig.reader_file()?;
            let _headers = rdr.headers()?.clone();
            return count_chunk_outliers(&fields_to_count, rdr.byte_records());
        }

        let njobs = util::njobs(flag_jobs);
        let chunk_size = util::chunk_size(idx_count, njobs);
        let nchunks = util::num_of_chunks(idx_count, chunk_size);

        log::info!("Parallelizing outlier counting: {nchunks} chunks, {njobs} jobs");

        let pool = ThreadPool::new(njobs);
        let (send, recv) = crossbeam_channel::bounded(nchunks);

        // Process each chunk in parallel. Share the read-only field map via Arc
        // instead of deep-cloning the HashMap into every worker. Since the caller
        // no longer needs `fields_to_count` after this call, we move it directly
        // into the Arc — zero map clones. `input_path_string` from above is already
        // UTF-8-validated, so no need to re-validate here.
        let fields_arc = Arc::new(fields_to_count);
        for i in 0..nchunks {
            let send = send.clone();
            let fields_arc = Arc::clone(&fields_arc);
            let input_path_string_clone = input_path_string.clone();
            pool.execute(move || {
                // Open index for this thread. If this fails, propagate an error
                // through the channel — dropping the chunk silently would
                // under-count outliers without any indication to the user.
                let rconfig_chunk = Config::new(Some(&input_path_string_clone));
                let mut idx_chunk = match rconfig_chunk.indexed() {
                    Ok(Some(idx)) => idx,
                    Ok(None) => {
                        let _ = send.send(Err(CliError::Other(format!(
                            "Chunk {i}: index is not available for {input_path_string_clone}"
                        ))));
                        return;
                    },
                    Err(e) => {
                        let _ = send.send(Err(CliError::Other(format!(
                            "Chunk {i}: failed to open index: {e}"
                        ))));
                        return;
                    },
                };

                // Seek to chunk start position
                if let Err(e) = idx_chunk.seek((i * chunk_size) as u64) {
                    let _ = send.send(Err(CliError::Other(format!("Chunk {i}: seek failed: {e}"))));
                    return;
                }

                // Process chunk records
                let it = idx_chunk.byte_records().take(chunk_size);
                let result = count_chunk_outliers(&fields_arc, it);
                let _ = send.send(result);
            });
        }

        drop(send);

        // Aggregate results from all chunks
        let mut all_stats: HashMap<String, OutlierStats> = fields_arc
            .keys()
            .map(|k| (k.clone(), OutlierStats::default()))
            .collect();

        for chunk_result in &recv {
            let chunk_stats = chunk_result?;
            for (field_name, stats) in chunk_stats {
                if let Some(total_stats) = all_stats.get_mut(&field_name) {
                    // Aggregate counts
                    for i in 0..OUTLIER_COUNTS_LEN {
                        total_stats.counts[i] += stats.counts[i];
                    }
                    // Aggregate sums
                    total_stats.sum_outliers += stats.sum_outliers;
                    total_stats.sum_normal += stats.sum_normal;
                    total_stats.sum_all += stats.sum_all;
                    total_stats.count_all += stats.count_all;
                    // Aggregate winsorized/trimmed stats
                    total_stats.winsorized_sum += stats.winsorized_sum;
                    total_stats.winsorized_count += stats.winsorized_count;
                    total_stats.trimmed_sum += stats.trimmed_sum;
                    total_stats.trimmed_count += stats.trimmed_count;
                    // Aggregate sum of squares
                    total_stats.sum_squares_outliers += stats.sum_squares_outliers;
                    total_stats.sum_squares_normal += stats.sum_squares_normal;
                    total_stats.sum_squares_trimmed += stats.sum_squares_trimmed;
                    total_stats.sum_squares_winsorized += stats.sum_squares_winsorized;
                    // Aggregate min/max
                    if let Some(min) = stats.min_outliers {
                        total_stats.min_outliers =
                            Some(total_stats.min_outliers.map_or(min, |m| m.min(min)));
                    }
                    if let Some(max) = stats.max_outliers {
                        total_stats.max_outliers =
                            Some(total_stats.max_outliers.map_or(max, |m| m.max(max)));
                    }
                    if let Some(min) = stats.min_normal {
                        total_stats.min_normal =
                            Some(total_stats.min_normal.map_or(min, |m| m.min(min)));
                    }
                    if let Some(max) = stats.max_normal {
                        total_stats.max_normal =
                            Some(total_stats.max_normal.map_or(max, |m| m.max(max)));
                    }
                    if let Some(min) = stats.min_trimmed {
                        total_stats.min_trimmed =
                            Some(total_stats.min_trimmed.map_or(min, |m| m.min(min)));
                    }
                    if let Some(max) = stats.max_trimmed {
                        total_stats.max_trimmed =
                            Some(total_stats.max_trimmed.map_or(max, |m| m.max(max)));
                    }
                    if let Some(min) = stats.min_winsorized {
                        total_stats.min_winsorized =
                            Some(total_stats.min_winsorized.map_or(min, |m| m.min(min)));
                    }
                    if let Some(max) = stats.max_winsorized {
                        total_stats.max_winsorized =
                            Some(total_stats.max_winsorized.map_or(max, |m| m.max(max)));
                    }
                }
            }
        }

        Ok(all_stats)
    } else {
        // Sequential fallback when no index exists
        let mut rdr = rconfig.reader_file()?;
        let _headers = rdr.headers()?.clone();
        count_chunk_outliers(&fields_to_count, rdr.byte_records())
    }
}

/// Process a chunk of records and update bivariate statistics
/// Similar to count_chunk_outliers but for bivariate computation
fn compute_chunk_bivariate<I>(
    field_pairs: &HashMap<(u16, u16), (BivariateFieldInfo, BivariateFieldInfo)>,
    records: I,
    stats_config: BivariateStatsConfig,
) -> CliResult<HashMap<(u16, u16), BivariateChunkStats>>
where
    I: Iterator<Item = csv::Result<csv::ByteRecord>>,
{
    if field_pairs.is_empty() {
        return Ok(HashMap::new());
    }

    // Check what we need to compute based on config
    let needs_all_values = stats_config.needs_all_values();
    let needs_freq_counts = stats_config.needs_frequency_counts();

    // Initialize statistics for all field pairs
    // Pre-allocate vectors with estimated capacity (typical chunk size is 1k-10k records)
    let estimated_capacity = 5000; // Reasonable estimate for chunk processing
    let estimated_unique_strings = estimated_capacity.min(1000); // Estimate for string frequency maps
    let mut chunk_stats: HashMap<(u16, u16), BivariateChunkStats> = field_pairs
        .keys()
        .map(|k| {
            let mut stats = BivariateChunkStats::default();
            // Only allocate value vectors if needed for Spearman/Kendall
            if needs_all_values {
                stats.x_values.reserve(estimated_capacity);
                stats.y_values.reserve(estimated_capacity);
            }
            // Only allocate frequency maps if needed for mutual information
            if needs_freq_counts {
                stats.xy_counts.reserve(estimated_unique_strings);
                stats.x_counts.reserve(estimated_unique_strings / 2);
                stats.y_counts.reserve(estimated_unique_strings / 2);
            }
            (*k, stats)
        })
        .collect();

    let prefer_dmy = util::get_envvar_flag("QSV_PREFER_DMY");

    // Optimization #1: Date parsing cache - Cache parsed dates to avoid re-parsing same strings
    let mut date_cache: HashMap<String, Option<f64>> = HashMap::with_capacity(estimated_capacity);

    // Optimization #6: String interning - Cache frequently used strings to reduce allocations.
    // Only needed if we're computing mutual information. A HashSet suffices since key==value;
    // saves one clone per cache miss over the prior HashMap<String, String> form.
    let mut string_interner: HashSet<String> = if needs_freq_counts {
        HashSet::with_capacity(estimated_unique_strings)
    } else {
        HashSet::new()
    };

    #[allow(unused_assignments)]
    let mut record: csv::ByteRecord = csv::ByteRecord::new();

    // Process each record in the chunk
    for result in records {
        record = result?;

        for ((idx1, idx2), (field1_info, field2_info)) in field_pairs {
            let value_bytes_x = record.get(field1_info.col_idx).unwrap_or(&[]);
            let value_bytes_y = record.get(field2_info.col_idx).unwrap_or(&[]);

            if value_bytes_x.is_empty() || value_bytes_y.is_empty() {
                continue;
            }

            // safety: chunk_stats is pre-populated with all field pair indices
            let Some(stats) = chunk_stats.get_mut(&(*idx1, *idx2)) else {
                debug_assert!(false, "chunk_stats missing expected key: ({idx1}, {idx2})");
                continue;
            };

            // Date parsing needs &str; numeric parsing reads bytes directly to avoid an
            // allocation on every record. Per-cell from_utf8 is a cheap byte scan compared
            // to the String allocation it replaces, so we no longer pre-stringify needed
            // columns into a per-record HashMap.
            let numeric_value_x = if field1_info.field_type.is_date_or_datetime() {
                let Ok(x_str) = from_utf8(value_bytes_x) else {
                    continue;
                };
                if let Some(cached) = date_cache.get(x_str) {
                    *cached
                } else {
                    let val = parse_date_to_days(x_str, prefer_dmy);
                    date_cache.insert(x_str.to_string(), val);
                    val
                }
            } else {
                parse_float_opt_from_bytes(value_bytes_x)
            };

            let numeric_value_y = if field2_info.field_type.is_date_or_datetime() {
                let Ok(y_str) = from_utf8(value_bytes_y) else {
                    continue;
                };
                if let Some(cached) = date_cache.get(y_str) {
                    *cached
                } else {
                    let val = parse_date_to_days(y_str, prefer_dmy);
                    date_cache.insert(y_str.to_string(), val);
                    val
                }
            } else {
                parse_float_opt_from_bytes(value_bytes_y)
            };

            // For numeric/date types, update correlation state and collect values
            if let (Some(x_val), Some(y_val)) = (numeric_value_x, numeric_value_y) {
                update_correlation_state(&mut stats.correlation_state, x_val, y_val);
                // Only store values if needed for Spearman/Kendall
                if needs_all_values {
                    stats.x_values.push(x_val);
                    stats.y_values.push(y_val);
                }
            }

            // Only compute frequency counts if needed for mutual information.
            // This is the only branch that needs owned strings — allocate lazily here.
            if needs_freq_counts {
                let (Ok(x_str_ref), Ok(y_str_ref)) =
                    (from_utf8(value_bytes_x), from_utf8(value_bytes_y))
                else {
                    continue;
                };
                // NOTE: this is not true string interning — each lookup still yields
                // a cloned String (needed because xy_counts keys by owned tuples).
                // The benefit over ad-hoc cloning is one fewer clone on insert (cache
                // miss). A future change to Arc<str> or SmolStr would give real
                // interning (cheap clones on cache hit), at the cost of a wider
                // refactor of xy_counts.
                let x_str_interned = if let Some(cached) = string_interner.get(x_str_ref) {
                    cached.clone()
                } else {
                    let owned = x_str_ref.to_string();
                    string_interner.insert(owned.clone());
                    owned
                };
                let y_str_interned = if let Some(cached) = string_interner.get(y_str_ref) {
                    cached.clone()
                } else {
                    let owned = y_str_ref.to_string();
                    string_interner.insert(owned.clone());
                    owned
                };

                // Accumulate joint frequency counts (xy_counts) - these are needed for mutual
                // information. Marginal frequencies (x_counts, y_counts) will be computed
                // from xy_counts at finalization to ensure consistency.
                *stats
                    .xy_counts
                    .entry((x_str_interned, y_str_interned))
                    .or_insert(0) += 1;
                stats.total_pairs += 1;
            }
        }
    }

    Ok(chunk_stats)
}

/// Finalize per-pair bivariate statistics from an aggregated `BivariateChunkStats`.
///
/// Caller must have already populated marginal frequencies (`x_counts`/`y_counts`)
/// from `xy_counts` when mutual information / NMI are requested. This function only
/// reads from `chunk_stats` — it does not mutate frequency maps.
fn finalize_bivariate_pair_stats(
    pair_key: (u16, u16),
    chunk_stats: &BivariateChunkStats,
    field_pairs: &HashMap<(u16, u16), (BivariateFieldInfo, BivariateFieldInfo)>,
    field_names: &[String],
    cardinality_threshold: Option<u64>,
    stats_config: BivariateStatsConfig,
) -> CliResult<((u16, u16), BivariateStats)> {
    let n_pairs = chunk_stats
        .correlation_state
        .count
        .max(chunk_stats.total_pairs);

    // chunk_stats keys mirror field_pairs keys; a miss indicates an invariant violation.
    let (field1_info, field2_info) = field_pairs.get(&pair_key).ok_or_else(|| {
        CliError::Other(format!(
            "Invariant violation: field pair not found: {pair_key:?}"
        ))
    })?;

    // Early exit: skip all correlation/covariance computations if variance is zero
    let has_zero_variance = field1_info.stddev.is_some_and(|s| s.abs() < f64::EPSILON)
        || field2_info.stddev.is_some_and(|s| s.abs() < f64::EPSILON)
        || field1_info.variance.is_some_and(|v| v.abs() < f64::EPSILON)
        || field2_info.variance.is_some_and(|v| v.abs() < f64::EPSILON);

    let pearson =
        if !stats_config.pearson || has_zero_variance || chunk_stats.correlation_state.count < 2 {
            None
        } else {
            finalize_pearson_correlation(&chunk_stats.correlation_state)
        };

    let (covariance_sample, covariance_population) =
        if !stats_config.covariance || has_zero_variance || chunk_stats.correlation_state.count < 2
        {
            (None, None)
        } else {
            (
                finalize_covariance(&chunk_stats.correlation_state, true),
                finalize_covariance(&chunk_stats.correlation_state, false),
            )
        };

    let spearman = if !stats_config.spearman || has_zero_variance || chunk_stats.x_values.len() < 2
    {
        None
    } else {
        compute_spearman_correlation(&chunk_stats.x_values, &chunk_stats.y_values)
    };

    let kendall = if !stats_config.kendall || has_zero_variance || chunk_stats.x_values.len() < 2 {
        None
    } else {
        compute_kendall_tau(&chunk_stats.x_values, &chunk_stats.y_values)
    };

    // MI / NMI share a cardinality-threshold gate; compute it once.
    // `exceeds_cardinality` is None when no threshold is configured (always proceed).
    let exceeds_cardinality = cardinality_threshold.map(|threshold| {
        field1_info.cardinality.is_some_and(|c| c > threshold)
            || field2_info.cardinality.is_some_and(|c| c > threshold)
    });

    let log_skip = |what: &str| {
        if let Some(threshold) = cardinality_threshold {
            let (idx1, idx2) = pair_key;
            let field1_name = field_names
                .get(idx1 as usize)
                .map_or("?", std::string::String::as_str);
            let field2_name = field_names
                .get(idx2 as usize)
                .map_or("?", std::string::String::as_str);
            log::debug!(
                "Skipping {what} for pair ({field1_name}, {field2_name}) - cardinality exceeds \
                 threshold {threshold}"
            );
        }
    };

    let mutual_information = if !stats_config.mi || chunk_stats.total_pairs == 0 {
        None
    } else if exceeds_cardinality == Some(true) {
        log_skip("mutual information");
        None
    } else {
        compute_mutual_information_from_counts(
            &chunk_stats.xy_counts,
            &chunk_stats.x_counts,
            &chunk_stats.y_counts,
            chunk_stats.total_pairs,
        )
    };

    // NMI requires MI and entropies computed from the same frequency counts.
    let normalized_mutual_information = if !stats_config.nmi || chunk_stats.total_pairs == 0 {
        None
    } else if exceeds_cardinality == Some(true) {
        log_skip("normalized mutual information");
        None
    } else {
        let h_x = compute_entropy_from_counts(&chunk_stats.x_counts, chunk_stats.total_pairs);
        let h_y = compute_entropy_from_counts(&chunk_stats.y_counts, chunk_stats.total_pairs);
        let mi = if mutual_information.is_some() {
            mutual_information
        } else {
            compute_mutual_information_from_counts(
                &chunk_stats.xy_counts,
                &chunk_stats.x_counts,
                &chunk_stats.y_counts,
                chunk_stats.total_pairs,
            )
        };
        compute_normalized_mutual_information(mi, h_x, h_y)
    };

    Ok((
        pair_key,
        BivariateStats {
            pearson,
            spearman,
            kendall,
            covariance_sample,
            covariance_population,
            mutual_information,
            normalized_mutual_information,
            n_pairs,
        },
    ))
}

/// Compute all bivariate statistics
/// Uses parallel chunked processing when an index is available and there
/// are more than 10,000 records.
/// Otherwise, uses sequential processing.
/// Returns a HashMap mapping field pairs to their bivariate statistics.
fn compute_all_bivariatestats(
    field_pairs: HashMap<(u16, u16), (BivariateFieldInfo, BivariateFieldInfo)>,
    field_names: &[String],
    input_path: &Path,
    progress: Option<&ProgressBar>,
    cardinality_threshold: Option<u64>,
    stats_config: BivariateStatsConfig,
    flag_jobs: Option<usize>,
) -> CliResult<HashMap<(u16, u16), BivariateStats>> {
    if field_pairs.is_empty() {
        return Ok(HashMap::new());
    }

    // Check what we need based on config
    let needs_all_values = stats_config.needs_all_values();
    let needs_freq_counts = stats_config.needs_frequency_counts();

    // Check if index exists for parallel processing
    let input_path_str = input_path
        .to_str()
        .ok_or_else(|| CliError::Other(format!("Invalid input path: {}", input_path.display())))?;
    let input_path_string = input_path_str.to_string();
    let rconfig = Config::new(Some(&input_path_string));
    let indexed_result = rconfig.indexed()?;

    if let Some(idx) = indexed_result {
        // Parallel processing path
        let idx_count = idx.count() as usize;
        if idx_count == 0 {
            return Ok(HashMap::new());
        }

        // Only parallelize if file is large enough
        if idx_count < PARALLEL_THRESHOLD {
            // Fall back to sequential for small files
            let mut rdr = rconfig.reader_file()?;
            let _headers = rdr.headers()?.clone();
            winfo!("Computing bivariate statistics sequentially...");
            return compute_all_bivariatestats_sequential(
                &field_pairs,
                field_names,
                rdr,
                progress,
                cardinality_threshold,
                stats_config,
            );
        }

        let njobs = util::njobs(flag_jobs);
        let chunk_size = util::chunk_size(idx_count, njobs);
        let nchunks = util::num_of_chunks(idx_count, chunk_size);

        winfo!("Parallelizing bivariate computation: {nchunks} chunks, {njobs} jobs");

        // Setup progress bar if requested
        if let Some(pb) = progress {
            pb.set_style(
                ProgressStyle::default_bar()
                    .template(
                        "[{elapsed_precise}] [{wide_bar} {percent}%{msg}] ({per_sec} - {eta})",
                    )
                    .unwrap(),
            );
            pb.set_message(format!(" of {} chunks", HumanCount(nchunks as u64)));
            pb.set_length(nchunks as u64);
            log::info!("Progress started... {nchunks} chunks");
        }

        let pool = ThreadPool::new(njobs);
        let (send, recv) = crossbeam_channel::bounded(nchunks);

        // Process each chunk in parallel. Share the read-only field-pair map via
        // Arc instead of deep-cloning the HashMap into every worker. The caller
        // no longer needs `field_pairs` after this call, so we move it directly
        // into the Arc — zero map clones. `input_path_string` from above is already
        // UTF-8-validated, so no need to re-validate here.
        let field_pairs_arc = Arc::new(field_pairs);
        for i in 0..nchunks {
            let send = send.clone();
            let field_pairs_arc = Arc::clone(&field_pairs_arc);
            let input_path_string_clone = input_path_string.clone();
            pool.execute(move || {
                // Open index for this thread. If this fails, propagate an error
                // through the channel — dropping the chunk silently would
                // produce incorrect bivariate stats without any indication.
                let rconfig_chunk = Config::new(Some(&input_path_string_clone));
                let mut idx_chunk = match rconfig_chunk.indexed() {
                    Ok(Some(idx)) => idx,
                    Ok(None) => {
                        let _ = send.send(Err(CliError::Other(format!(
                            "Chunk {i}: index is not available for {input_path_string_clone}"
                        ))));
                        return;
                    },
                    Err(e) => {
                        let _ = send.send(Err(CliError::Other(format!(
                            "Chunk {i}: failed to open index: {e}"
                        ))));
                        return;
                    },
                };

                // Seek to chunk start position
                if let Err(e) = idx_chunk.seek((i * chunk_size) as u64) {
                    let _ = send.send(Err(CliError::Other(format!("Chunk {i}: seek failed: {e}"))));
                    return;
                }

                // Process chunk records
                let it = idx_chunk.byte_records().take(chunk_size);
                let result = compute_chunk_bivariate(&field_pairs_arc, it, stats_config);
                let _ = send.send(result);
            });
        }

        drop(send);

        // Aggregate results from all chunks
        // Pre-allocate based on idx_count to avoid repeated reallocations during extend
        let mut all_stats: HashMap<(u16, u16), BivariateChunkStats> = field_pairs_arc
            .keys()
            .map(|k| {
                let mut stats = BivariateChunkStats::default();
                // Pre-allocate value vectors with total capacity if needed
                if needs_all_values {
                    stats.x_values.reserve(idx_count);
                    stats.y_values.reserve(idx_count);
                }
                (*k, stats)
            })
            .collect();

        for chunk_result in &recv {
            let chunk_stats = chunk_result?;
            for (pair_key, stats) in chunk_stats {
                if let Some(total_stats) = all_stats.get_mut(&pair_key) {
                    // Merge correlation states (always needed for Pearson/covariance)
                    total_stats.correlation_state = merge_correlation_states(
                        &total_stats.correlation_state,
                        &stats.correlation_state,
                    );
                    // Only merge values if needed for Spearman/Kendall
                    if needs_all_values {
                        total_stats.x_values.extend(stats.x_values);
                        total_stats.y_values.extend(stats.y_values);
                    }
                    // Only merge frequency counts if needed for mutual information
                    // Note: Only xy_counts and total_pairs are collected during chunk processing
                    // Marginal frequencies (x_counts, y_counts) are computed from xy_counts at
                    // finalization
                    if needs_freq_counts {
                        for ((x_val, y_val), count) in stats.xy_counts {
                            *total_stats.xy_counts.entry((x_val, y_val)).or_insert(0) += count;
                        }
                        total_stats.total_pairs += stats.total_pairs;
                    }
                }
            }
            // Update progress bar
            if let Some(pb) = progress {
                pb.inc(1);
            }
        }

        winfo!("Finalizing bivariate statistics...");
        // Update progress bar for Phase 2: final statistics computation
        let num_field_pairs = all_stats.len();
        if let Some(pb) = progress {
            pb.set_style(
                ProgressStyle::default_bar()
                    .template(
                        "[{elapsed_precise}] [{wide_bar} {percent}%{msg}] ({per_sec} - {eta})",
                    )
                    .unwrap(),
            );
            pb.set_message(format!(
                " of {} field pairs",
                HumanCount(num_field_pairs as u64)
            ));
            pb.set_length(num_field_pairs as u64);
            pb.set_position(0); // Reset position for Phase 2
            log::info!("Phase 2 started... {num_field_pairs} field pairs");
        }

        // Only compute marginal frequencies if we need mutual information
        if needs_freq_counts {
            // Compute marginal frequencies from joint frequencies to ensure consistency
            // This ensures x_counts and y_counts are computed from the same set of records
            // as xy_counts (only pairs where both fields are non-empty)
            // This is critical for correct mutual information calculation
            for chunk_stats in all_stats.values_mut() {
                // Compute marginal frequencies from joint frequencies
                // Sum over y to get x_counts, sum over x to get y_counts
                chunk_stats.x_counts.clear();
                chunk_stats.y_counts.clear();

                for ((x_val, y_val), &count) in &chunk_stats.xy_counts {
                    *chunk_stats.x_counts.entry(x_val.clone()).or_insert(0) += count;
                    *chunk_stats.y_counts.entry(y_val.clone()).or_insert(0) += count;
                }
            }
        }

        // Finalize statistics from aggregated chunk stats (parallelized)
        let final_stats: HashMap<(u16, u16), BivariateStats> = all_stats
            .into_par_iter()
            .map(|(pair_key, chunk_stats)| {
                if let Some(pb) = progress {
                    pb.inc(1);
                }
                finalize_bivariate_pair_stats(
                    pair_key,
                    &chunk_stats,
                    &field_pairs_arc,
                    field_names,
                    cardinality_threshold,
                    stats_config,
                )
            })
            .collect::<CliResult<HashMap<_, _>>>()?;

        // Finish progress bar after final statistics computation
        if let Some(pb) = progress {
            util::finish_progress(pb);
        }

        Ok(final_stats)
    } else {
        // Sequential fallback when no index exists
        let mut rdr = rconfig.reader_file()?;
        let _headers = rdr.headers()?.clone();
        compute_all_bivariatestats_sequential(
            &field_pairs,
            field_names,
            rdr,
            progress,
            cardinality_threshold,
            stats_config,
        )
    }
}

/// Sequential processing for small files (< 10k records) or when no index exists.
///
/// Delegates to `compute_chunk_bivariate` for the per-record scan (treating the whole
/// file as a single chunk), then computes marginal frequencies and finalizes per pair
/// via `finalize_bivariate_pair_stats` — sharing all of that logic with the parallel path.
fn compute_all_bivariatestats_sequential(
    field_pairs: &HashMap<(u16, u16), (BivariateFieldInfo, BivariateFieldInfo)>,
    field_names: &[String],
    mut rdr: csv::Reader<std::fs::File>,
    progress: Option<&ProgressBar>,
    cardinality_threshold: Option<u64>,
    stats_config: BivariateStatsConfig,
) -> CliResult<HashMap<(u16, u16), BivariateStats>> {
    if field_pairs.is_empty() {
        return Ok(HashMap::new());
    }

    let needs_freq_counts = stats_config.needs_frequency_counts();

    // Set up progress bar once before iteration (unknown total, ticks per record).
    if let Some(pb) = progress {
        pb.set_style(
            ProgressStyle::default_bar()
                .template("[{elapsed_precise}] [{wide_bar}] {pos} records ({per_sec})")
                .unwrap(),
        );
        pb.set_length(0);
    }

    // Wrap byte_records() with inspect() so the progress bar ticks every 1000 records
    // without coupling compute_chunk_bivariate to a ProgressBar parameter.
    let mut processed = 0u64;
    let it = rdr.byte_records().inspect(|_| {
        processed += 1;
        if let Some(pb) = progress
            && processed.is_multiple_of(1000)
        {
            pb.set_position(processed);
        }
    });
    let mut all_stats = compute_chunk_bivariate(field_pairs, it, stats_config)?;

    if let Some(pb) = progress {
        pb.set_position(processed);
        util::finish_progress(pb);
    }

    // Compute marginal frequencies from joint frequencies (same as parallel path).
    if needs_freq_counts {
        for chunk_stats in all_stats.values_mut() {
            chunk_stats.x_counts.clear();
            chunk_stats.y_counts.clear();
            for ((x_val, y_val), &count) in &chunk_stats.xy_counts {
                *chunk_stats.x_counts.entry(x_val.clone()).or_insert(0) += count;
                *chunk_stats.y_counts.entry(y_val.clone()).or_insert(0) += count;
            }
        }
    }

    all_stats
        .into_iter()
        .map(|(pair_key, chunk_stats)| {
            finalize_bivariate_pair_stats(
                pair_key,
                &chunk_stats,
                field_pairs,
                field_names,
                cardinality_threshold,
                stats_config,
            )
        })
        .collect()
}

/// Compute Kurtosis, Gini coefficient, and Atkinson index for all fields.
/// Since Kurtosis, Gini & Atkinson Index require all values from the entire dataset, this always
/// uses sequential processing to read all values in a single pass.
/// Returns a HashMap mapping field names to their Kurtosis, Gini coefficient, and Atkinson index
/// statistics
fn compute_all_kga(
    fields_to_compute: &HashMap<String, KGAFieldInfo>,
    input_path: &Path,
    atkinson_epsilon: f64,
) -> CliResult<HashMap<String, KGAStats>> {
    if fields_to_compute.is_empty() {
        return Ok(HashMap::new());
    }

    let input_path_str = input_path
        .to_str()
        .ok_or_else(|| CliError::Other(format!("Invalid input path: {}", input_path.display())))?;
    let input_path_string = input_path_str.to_string();
    let rconfig = Config::new(Some(&input_path_string));
    let mut rdr = rconfig.reader_file()?;
    let _headers = rdr.headers()?.clone();
    compute_all_kga_from_reader(fields_to_compute, rdr, atkinson_epsilon)
}

/// Compute Kurtosis, Gini coefficient, and Atkinson index for all fields in a single pass through
/// the CSV (sequential) The CSV reader should already be positioned after the headers
/// Returns a HashMap mapping field names to their Kurtosis, Gini coefficient, and Atkinson index
/// statistics
fn compute_all_kga_from_reader(
    fields_to_compute: &HashMap<String, KGAFieldInfo>,
    mut rdr: csv::Reader<std::fs::File>,
    atkinson_epsilon: f64,
) -> CliResult<HashMap<String, KGAStats>> {
    if fields_to_compute.is_empty() {
        return Ok(HashMap::new());
    }

    // Collect all values for each field
    let mut field_values: HashMap<String, Vec<f64>> = fields_to_compute
        .keys()
        .map(|k| (k.clone(), Vec::new()))
        .collect();

    let prefer_dmy = util::get_envvar_flag("QSV_PREFER_DMY");

    // amortize allocations
    #[allow(unused_assignments)]
    let mut record: StringRecord = StringRecord::new();
    let mut value_str;
    let mut numeric_value;

    // Process each record once, collecting values for all fields
    for result in rdr.records() {
        record = result?;

        for (field_name, field_info) in fields_to_compute {
            value_str = record.get(field_info.col_idx).unwrap_or("");

            if value_str.is_empty() {
                continue; // Skip null/empty values
            }

            // Parse the value based on field type
            numeric_value = if field_info.field_type.is_date_or_datetime() {
                parse_date_to_days(value_str, prefer_dmy)
            } else {
                parse_float_opt(value_str)
            };

            if let Some(val) = numeric_value
                && let Some(values) = field_values.get_mut(field_name)
            {
                values.push(val);
            }
        }
    }

    // Compute statistics for each field
    let mut all_stats: HashMap<String, KGAStats> = HashMap::with_capacity(field_values.len());

    for (field_name, values) in field_values {
        if values.len() < 2 {
            // Need at least 2 values for meaningful statistics
            all_stats.insert(
                field_name,
                KGAStats {
                    kurtosis:         None,
                    gini_coefficient: None,
                    atkinson_index:   None,
                    theil_index:      None,
                    mean_ad:          None,
                },
            );
            continue;
        }

        // Get precalculated stats for this field
        let (precalc_mean, precalc_variance, precalc_sum) = fields_to_compute
            .get(&field_name)
            .map_or((None, None, None), |info| {
                (info.mean, info.variance, info.sum)
            });

        // Compute kurtosis with precalculated mean and variance
        let kurtosis_val = kurtosis(values.iter().copied(), precalc_mean, precalc_variance);

        // Compute Gini coefficient with precalculated sum (not mean!)
        let gini_val = gini(values.iter().copied(), precalc_sum);

        // Compute Atkinson Index (epsilon parameter typically 0.5 or 1.0, configurable via
        // --epsilon) atkinson function signature: atkinson(iter, epsilon,
        // precalc_mean, precalc_geometric_sum) See: https://docs.rs/qsv-stats/latest/stats/fn.atkinson.html
        let atkinson_val = atkinson(
            values.iter().copied(),
            atkinson_epsilon,
            precalc_mean,
            None, // geometric sum not precalculated
        );

        // Compute Theil Index: (1/n) * Σ((x_i / mean) * ln(x_i / mean))
        // Only for positive values (Theil index is undefined for non-positive values)
        // Uses mean of positive values only, so it works even when overall mean is <= 0
        #[allow(clippy::cast_precision_loss)]
        let theil_val = {
            // First pass: compute sum and count of positive values
            let mut pos_sum = 0.0_f64;
            let mut pos_count: usize = 0;
            for &v in &values {
                if v > 0.0 {
                    pos_sum += v;
                    pos_count += 1;
                }
            }

            if pos_count >= 2 {
                let n = pos_count as f64;
                let pos_mean = pos_sum / n;
                if pos_mean > f64::EPSILON {
                    // Second pass: accumulate Theil sum over positive values
                    let mut theil_sum = 0.0_f64;
                    for &v in &values {
                        if v > 0.0 {
                            let ratio = v / pos_mean;
                            // use CPU's Fused Multiply-Add (FMA) for better precision and to reduce
                            // intermediate rounding errors in the sum
                            // theil_sum += ratio * ratio.ln();
                            theil_sum = ratio.mul_add(ratio.ln(), theil_sum);
                        }
                    }
                    Some(theil_sum / n)
                } else {
                    None
                }
            } else {
                None
            }
        };

        // Compute Mean Absolute Deviation from mean: (1/n) * Σ|x_i - mean|
        #[allow(clippy::cast_precision_loss)]
        let mean_ad_val = if let Some(mean_val) = precalc_mean {
            let n = values.len() as f64;
            let sum_abs_dev: f64 = values.iter().map(|&x| (x - mean_val).abs()).sum();
            Some(sum_abs_dev / n)
        } else {
            None
        };

        all_stats.insert(
            field_name,
            KGAStats {
                kurtosis:         kurtosis_val,
                gini_coefficient: gini_val,
                atkinson_index:   atkinson_val,
                theil_index:      theil_val,
                mean_ad:          mean_ad_val,
            },
        );
    }

    Ok(all_stats)
}

/// Compute Shannon Entropy for all fields by calling the frequency command.
/// Uses run_qsv_cmd to call frequency command with --limit 0 to get all frequencies,
/// then parses the CSV output and computes entropy for each field.
/// Returns a HashMap mapping field names to their entropy statistics
fn compute_all_entropy(input_path: &Path) -> CliResult<HashMap<String, EntropyStats>> {
    let input_path_str = input_path
        .to_str()
        .ok_or_else(|| CliError::Other(format!("Invalid input path: {}", input_path.display())))?;

    // Call frequency command with --limit 0 to get all frequencies for all fields
    let (freq_output, _) = util::run_qsv_cmd(
        "frequency",
        &["--limit", "0"],
        input_path_str,
        "Computing frequency distributions for entropy...",
    )?;

    // Parse the frequency CSV output
    // Format: field,value,count,percentage,rank
    let mut rdr = ReaderBuilder::new()
        .has_headers(true)
        .from_reader(freq_output.as_bytes());

    let headers = rdr.headers()?.clone();
    let field_idx = headers
        .iter()
        .position(|h| h == "field")
        .ok_or_else(|| CliError::Other("Frequency CSV missing 'field' column".to_string()))?;
    let value_idx = headers
        .iter()
        .position(|h| h == "value")
        .ok_or_else(|| CliError::Other("Frequency CSV missing 'value' column".to_string()))?;
    let count_idx = headers
        .iter()
        .position(|h| h == "count")
        .ok_or_else(|| CliError::Other("Frequency CSV missing 'count' column".to_string()))?;

    // Group frequencies by field name
    let mut field_frequencies: HashMap<String, HashMap<String, u64>> = HashMap::new();
    let mut field_totals: HashMap<String, u64> = HashMap::new();

    for result in rdr.records() {
        let record = result?;
        let field_name = record.get(field_idx).unwrap_or("").to_string();
        let value = record.get(value_idx).unwrap_or("").to_string();
        let count: u64 = record
            .get(count_idx)
            .ok_or_else(|| CliError::Other("Missing count in frequency CSV".to_string()))?
            .parse()
            .map_err(|e| CliError::Other(format!("Failed to parse count: {e}")))?;

        // Skip empty field names (shouldn't happen, but be safe)
        if field_name.is_empty() {
            continue;
        }

        // Initialize field entry if needed
        field_frequencies
            .entry(field_name.clone())
            .or_default()
            .insert(value, count);

        // Accumulate total count for this field
        *field_totals.entry(field_name).or_insert(0) += count;
    }

    // Compute entropy for each field
    let mut entropy_stats: HashMap<String, EntropyStats> = HashMap::new();

    #[allow(clippy::cast_precision_loss)]
    for (field_name, frequencies) in field_frequencies {
        let total_count = field_totals.get(&field_name).copied().unwrap_or(0);

        if total_count == 0 {
            entropy_stats.insert(
                field_name,
                EntropyStats {
                    entropy:            None,
                    simpsons_diversity: None,
                },
            );
            continue;
        }

        // Check if this is an all-unique field (frequency command outputs <ALL_UNIQUE> for these)
        // The default text is "<ALL_UNIQUE>" but it can be customized with --all-unique-text
        // We check for both the default and common variations
        let is_all_unique = frequencies.len() == 1
            && frequencies.keys().any(|v| {
                v == "<ALL_UNIQUE>"
                    || v == "<ALL UNIQUE>"
                    || (v.starts_with("<ALL") && v.contains("UNIQUE"))
            });

        let (entropy, simpsons) = if is_all_unique {
            // For all-unique fields, each value appears exactly once
            // Entropy = log2(n) where n is the number of unique values (which equals total_count)
            // Formula: -Σ p_i * log2(p_i) where p_i = 1/n for each of n values
            // = -n * (1/n) * log2(1/n) = -log2(1/n) = log2(n)
            let entropy = (total_count as f64).log2();
            // Simpson's: 1 - Σ(p_i²) = 1 - n*(1/n)² = 1 - 1/n
            let simpsons = 1.0 - 1.0 / total_count as f64;
            (entropy, simpsons)
        } else {
            // Compute Shannon Entropy: H(X) = -Σ p_i * log2(p_i)
            // and Simpson's Diversity: 1 - Σ(p_i²)
            let mut entropy = 0.0;
            let mut sum_p_squared = 0.0;
            let total = total_count as f64;

            for count in frequencies.values() {
                if *count > 0 {
                    let p = *count as f64 / total;
                    entropy -= p * p.log2();
                    sum_p_squared += p * p;
                }
            }
            (entropy, 1.0 - sum_p_squared)
        };

        entropy_stats.insert(
            field_name,
            EntropyStats {
                entropy:            Some(entropy),
                simpsons_diversity: Some(simpsons),
            },
        );
    }

    Ok(entropy_stats)
}

pub fn run(argv: &[&str]) -> CliResult<()> {
    let start_time = Instant::now();
    let args: Args = util::get_args(USAGE, argv)?;

    // Read environment variables once at the top to avoid repeated reads in hot loops
    let stats_separator = std::env::var("QSV_STATS_SEPARATOR").unwrap_or_else(|_| "|".to_string());
    let prefer_dmy = util::get_envvar_flag("QSV_PREFER_DMY");

    // Check if input file is provided
    let input_path_str = args
        .arg_input
        .ok_or_else(|| CliError::IncorrectUsage("No input file specified.".to_string()))?;

    let input_path = Path::new(&input_path_str);
    if !input_path.exists() {
        return fail_clierror!("Input file does not exist: {}", input_path.display());
    }

    // Check atkinson epsilon is >= 0
    if args.flag_advanced && args.flag_epsilon < 0.0 {
        return fail_incorrectusage_clierror!(
            "Atkinson Index inequality aversion parameter must be >= 0. Got: {}",
            args.flag_epsilon
        );
    }

    // Handle multi-dataset join if requested
    let temp_joined_path: Option<PathBuf>;
    let actual_input_path = if let Some(ref join_inputs_str) = args.flag_join_inputs {
        let additional_inputs: Vec<String> = join_inputs_str
            .split(',')
            .map(|s| s.trim().to_string())
            .collect();
        let join_keys_str = args.flag_join_keys.as_ref().ok_or_else(|| {
            CliError::IncorrectUsage(
                "--join-keys required when --join-inputs is specified".to_string(),
            )
        })?;
        let join_keys: Vec<String> = join_keys_str
            .split(',')
            .map(|s| s.trim().to_string())
            .collect();
        let join_type = args.flag_join_type.as_deref().unwrap_or("inner");

        let joined_path =
            join_datasets_internal(input_path, &additional_inputs, &join_keys, join_type)?;
        temp_joined_path = Some(joined_path);
        temp_joined_path.as_ref().unwrap()
    } else {
        temp_joined_path = None;
        input_path
    };

    // Auto-create index if --advanced or --bivariate is set and index doesn't exist
    if args.flag_advanced || args.flag_bivariate {
        let actual_input_path_str = actual_input_path
            .to_str()
            .ok_or_else(|| CliError::Other("Invalid input path".to_string()))?
            .to_string();
        let rconfig = Config::new(Some(&actual_input_path_str));
        let indexed_result = rconfig.indexed()?;

        if indexed_result.is_none() && !rconfig.is_stdin() {
            let option_name = if args.flag_bivariate {
                "--bivariate"
            } else {
                "--advanced"
            };
            log::info!(
                "{option_name} option requires reading the entire CSV file. Auto-creating index \
                 to enable parallel processing..."
            );

            match util::create_index_for_file(actual_input_path, &rconfig) {
                Ok(()) => {
                    log::info!("Index created successfully for statistics computation.");
                },
                Err(index_err) => {
                    log::warn!("Failed to auto-create index: {index_err}");
                    // Continue anyway - the code will fall back to sequential processing
                },
            }
        }
    }

    // Determine stats CSV path
    // If we joined datasets, we need stats for the joined dataset, but write bivariate stats
    // based on the original input path
    let stats_csv_path = if temp_joined_path.is_some() {
        // For joined datasets, generate stats for the joined dataset
        // Use a temp stats CSV file
        let actual_input_path_str = actual_input_path
            .to_str()
            .ok_or_else(|| CliError::Other("Invalid joined path".to_string()))?
            .to_string();
        let temp_stats_file = tempfile::Builder::new().suffix(".stats.csv").tempfile_in(
            crate::config::TEMP_FILE_DIR.get_or_init(|| tempfile::TempDir::new().unwrap().keep()),
        )?;
        let temp_stats_path = temp_stats_file.path().to_path_buf();
        drop(temp_stats_file); // Close so stats can write to it

        // Generate stats for joined dataset
        let stats_args_vec: Vec<&str> = args.flag_stats_options.split_whitespace().collect();
        let mut stats_cmd_args = stats_args_vec.clone();
        stats_cmd_args.push("--output");
        let temp_stats_path_str = temp_stats_path
            .to_str()
            .ok_or_else(|| CliError::Other("Invalid temp stats path".to_string()))?;
        stats_cmd_args.push(temp_stats_path_str);

        let qsv_path = env::current_exe()
            .map_err(|e| CliError::Other(format!("Failed to get current executable path: {e:?}")))?
            .to_string_lossy()
            .to_string();
        let mut cmd = Command::new(&qsv_path);
        cmd.arg("stats")
            .args(&stats_cmd_args)
            .arg(&actual_input_path_str);
        let output = cmd
            .output()
            .map_err(|e| CliError::Other(format!("Error while executing stats command: {e:?}")))?;
        if !output.status.success() {
            return fail_clierror!(
                "Command stats failed: Output {{ status: {:?}, stdout: {:?}, stderr: {:?} }}",
                output.status,
                String::from_utf8_lossy(&output.stdout),
                String::from_utf8_lossy(&output.stderr)
            );
        }

        temp_stats_path
    } else {
        // For single dataset, use normal stats CSV path
        let path = get_stats_csv_path(input_path)?;

        // Check if stats CSV exists, if not, run stats command
        if args.flag_force || !path.exists() {
            if args.flag_force {
                winfo!("Force flag set: recomputing stats...");
            } else {
                wwarn!(
                    "Stats CSV file not found: {}\nComputing baseline stats...",
                    path.display()
                );
            }

            // Parse stats options
            let stats_args_vec: Vec<&str> = args.flag_stats_options.split_whitespace().collect();
            let _ = util::run_qsv_cmd(
                "stats",
                &stats_args_vec,
                &input_path_str,
                "Ran stats command to generate baseline stats...",
            )?;
            if !path.exists() {
                return fail_clierror!("Stats CSV file was not created: {}", path.display());
            }
        }

        path
    };

    // Read the stats CSV file
    let stats_csv_content = fs::read_to_string(&stats_csv_path)?;

    // Parse the stats CSV
    let mut rdr = ReaderBuilder::new()
        .has_headers(true)
        .from_reader(stats_csv_content.as_bytes());

    let headers = rdr.headers()?.clone();

    let type_idx = headers
        .iter()
        .position(|h| h == "type")
        .ok_or_else(|| CliError::Other("Stats CSV missing 'type' column".to_string()))?;

    let mean_idx = headers.iter().position(|h| h == "mean");
    let median_idx = headers.iter().position(|h| h == "median");
    let q2_median_idx = headers.iter().position(|h| h == "q2_median");
    let stddev_idx = headers.iter().position(|h| h == "stddev");
    let variance_idx = headers.iter().position(|h| h == "variance");
    let range_idx = headers.iter().position(|h| h == "range");
    let q1_idx = headers.iter().position(|h| h == "q1");
    let q3_idx = headers.iter().position(|h| h == "q3");
    let mode_idx = headers.iter().position(|h| h == "mode");
    let sem_idx = headers.iter().position(|h| h == "sem");
    let min_idx = headers.iter().position(|h| h == "min");
    let max_idx = headers.iter().position(|h| h == "max");
    let iqr_idx = headers.iter().position(|h| h == "iqr");
    let mad_idx = headers.iter().position(|h| h == "mad");
    let field_idx = headers.iter().position(|h| h == "field");
    let sum_idx = headers.iter().position(|h| h == "sum");
    let skewness_idx = headers.iter().position(|h| h == "skewness");
    let cardinality_idx = headers.iter().position(|h| h == "cardinality");
    let n_positive_idx = headers.iter().position(|h| h == "n_positive");
    let n_negative_idx = headers.iter().position(|h| h == "n_negative");
    let n_zero_idx = headers.iter().position(|h| h == "n_zero");
    let kurtosis_idx = headers.iter().position(|h| h == "kurtosis");
    let lower_outer_fence_idx = headers.iter().position(|h| h == "lower_outer_fence");
    let lower_inner_fence_idx = headers.iter().position(|h| h == "lower_inner_fence");
    let upper_inner_fence_idx = headers.iter().position(|h| h == "upper_inner_fence");
    let upper_outer_fence_idx = headers.iter().position(|h| h == "upper_outer_fence");
    let percentiles_idx = headers.iter().position(|h| h == "percentiles");

    // Parse and validate scan mode for Gregorian XSD date type detection
    let scan_mode = args.flag_xsd_gdate_scan.as_deref().unwrap_or("quick");
    if scan_mode != "quick" && scan_mode != "thorough" {
        return fail_clierror!(
            "Invalid scan mode: {}. Must be either 'quick' or 'thorough'",
            scan_mode
        );
    }

    // Parse and validate percentile thresholds if --use-percentiles is set
    let (lower_percentile, upper_percentile) = if args.flag_use_percentiles {
        let thresholds_str = args
            .flag_pct_thresholds
            .as_ref()
            .map_or("5,95", std::string::String::as_str);

        let parts: Vec<&str> = thresholds_str.split(',').map(str::trim).collect();
        if parts.len() != 2 {
            return fail_clierror!(
                "Invalid percentile thresholds: {}. Expected format: 'lower,upper' (e.g., '5,95')",
                thresholds_str
            );
        }

        let lower = fast_float2::parse::<f64, &[u8]>(parts[0].as_bytes()).map_err(|_| {
            CliError::IncorrectUsage(format!("Invalid lower percentile: {}", parts[0]))
        })?;
        let upper = fast_float2::parse::<f64, &[u8]>(parts[1].as_bytes()).map_err(|_| {
            CliError::IncorrectUsage(format!("Invalid upper percentile: {}", parts[1]))
        })?;

        if !(0.0..=100.0).contains(&lower) || !(0.0..=100.0).contains(&upper) {
            return fail_clierror!(
                "Percentile thresholds must be between 0 and 100. Got: {}, {}",
                lower,
                upper
            );
        }

        if lower >= upper {
            return fail_clierror!(
                "Lower percentile must be less than upper percentile. Got: {}, {}",
                lower,
                upper
            );
        }

        (Some(lower), Some(upper))
    } else {
        (None, None)
    };

    // Helper function to check if a column already exists in headers
    let column_exists = |col_name: &str| headers.iter().any(|h| h == col_name);

    // Generate Atkinson Index column name with epsilon parameter
    let atkinson_index_col_name = format!("atkinson_index_({})", args.flag_epsilon);

    // Check which new columns we can add (based on available base stats)
    // Skip columns that already exist to avoid duplicates
    let mut new_columns: Vec<String> = Vec::new();
    let mut new_column_indices = IndexMap::new();

    if mean_idx.is_some()
        && (median_idx.is_some() || q2_median_idx.is_some())
        && stddev_idx.is_some()
        && !column_exists("pearson_skewness")
    {
        new_columns.push("pearson_skewness".to_string());
        new_column_indices.insert("pearson_skewness".to_string(), new_columns.len() - 1);
    }

    if range_idx.is_some() && stddev_idx.is_some() && !column_exists("range_stddev_ratio") {
        new_columns.push("range_stddev_ratio".to_string());
        new_column_indices.insert("range_stddev_ratio".to_string(), new_columns.len() - 1);
    }

    if q1_idx.is_some() && q3_idx.is_some() && !column_exists("quartile_coefficient_dispersion") {
        new_columns.push("quartile_coefficient_dispersion".to_string());
        new_column_indices.insert(
            "quartile_coefficient_dispersion".to_string(),
            new_columns.len() - 1,
        );
    }

    if mode_idx.is_some()
        && mean_idx.is_some()
        && stddev_idx.is_some()
        && !column_exists("mode_zscore")
    {
        new_columns.push("mode_zscore".to_string());
        new_column_indices.insert("mode_zscore".to_string(), new_columns.len() - 1);
    }

    if sem_idx.is_some() && mean_idx.is_some() && !column_exists("relative_standard_error") {
        new_columns.push("relative_standard_error".to_string());
        new_column_indices.insert("relative_standard_error".to_string(), new_columns.len() - 1);
    }

    if min_idx.is_some()
        && mean_idx.is_some()
        && stddev_idx.is_some()
        && !column_exists("min_zscore")
    {
        new_columns.push("min_zscore".to_string());
        new_column_indices.insert("min_zscore".to_string(), new_columns.len() - 1);
    }

    if max_idx.is_some()
        && mean_idx.is_some()
        && stddev_idx.is_some()
        && !column_exists("max_zscore")
    {
        new_columns.push("max_zscore".to_string());
        new_column_indices.insert("max_zscore".to_string(), new_columns.len() - 1);
    }

    if (median_idx.is_some() || q2_median_idx.is_some())
        && mean_idx.is_some()
        && !column_exists("median_mean_ratio")
    {
        new_columns.push("median_mean_ratio".to_string());
        new_column_indices.insert("median_mean_ratio".to_string(), new_columns.len() - 1);
    }

    if iqr_idx.is_some() && range_idx.is_some() && !column_exists("iqr_range_ratio") {
        new_columns.push("iqr_range_ratio".to_string());
        new_column_indices.insert("iqr_range_ratio".to_string(), new_columns.len() - 1);
    }

    if mad_idx.is_some() && stddev_idx.is_some() && !column_exists("mad_stddev_ratio") {
        new_columns.push("mad_stddev_ratio".to_string());
        new_column_indices.insert("mad_stddev_ratio".to_string(), new_columns.len() - 1);
    }

    // Trimean: (Q1 + 2*median + Q3) / 4 - Tukey's robust central tendency estimator
    if q1_idx.is_some()
        && (median_idx.is_some() || q2_median_idx.is_some())
        && q3_idx.is_some()
        && !column_exists("trimean")
    {
        new_columns.push("trimean".to_string());
        new_column_indices.insert("trimean".to_string(), new_columns.len() - 1);
    }

    // Midhinge: (Q1 + Q3) / 2 - midpoint of the middle 50%
    if q1_idx.is_some() && q3_idx.is_some() && !column_exists("midhinge") {
        new_columns.push("midhinge".to_string());
        new_column_indices.insert("midhinge".to_string(), new_columns.len() - 1);
    }

    // Robust CV: MAD / median - outlier-resistant coefficient of variation
    if mad_idx.is_some()
        && (median_idx.is_some() || q2_median_idx.is_some())
        && !column_exists("robust_cv")
    {
        new_columns.push("robust_cv".to_string());
        new_column_indices.insert("robust_cv".to_string(), new_columns.len() - 1);
    }

    // Add kurtosis column (requires reading raw data, computed for numeric/date types)
    // Only add if --advanced flag is set
    if args.flag_advanced && !column_exists("kurtosis") {
        new_columns.push("kurtosis".to_string());
        new_column_indices.insert("kurtosis".to_string(), new_columns.len() - 1);
    }

    // Add bimodality coefficient (requires skewness from base stats and kurtosis from --advanced)
    // Only add if --advanced flag is set (since it requires kurtosis)
    if args.flag_advanced
        && skewness_idx.is_some()
        && new_column_indices.contains_key("kurtosis")
        && !column_exists("bimodality_coefficient")
    {
        new_columns.push("bimodality_coefficient".to_string());
        new_column_indices.insert("bimodality_coefficient".to_string(), new_columns.len() - 1);
    }

    // Add Jarque-Bera test statistic (requires skewness and kurtosis)
    // Only add if --advanced flag is set. Kurtosis can come from this run (new column)
    // or from a previous run (existing column in stats CSV).
    if args.flag_advanced
        && skewness_idx.is_some()
        && (new_column_indices.contains_key("kurtosis") || kurtosis_idx.is_some())
        && n_positive_idx.is_some()
        && n_negative_idx.is_some()
        && n_zero_idx.is_some()
        && !column_exists("jarque_bera")
    {
        new_columns.push("jarque_bera".to_string());
        new_column_indices.insert("jarque_bera".to_string(), new_columns.len() - 1);
        new_columns.push("jarque_bera_pvalue".to_string());
        new_column_indices.insert("jarque_bera_pvalue".to_string(), new_columns.len() - 1);
    }

    // Add Gini coefficient column (requires reading raw data, computed for numeric/date types)
    // Only add if --advanced flag is set
    if args.flag_advanced && !column_exists("gini_coefficient") {
        new_columns.push("gini_coefficient".to_string());
        new_column_indices.insert("gini_coefficient".to_string(), new_columns.len() - 1);
    }

    // Add Atkinson Index column (requires reading raw data, computed for numeric/date types)
    // Only add if --advanced flag is set
    if args.flag_advanced && !column_exists(&atkinson_index_col_name) {
        new_columns.push(atkinson_index_col_name.clone());
        new_column_indices.insert(atkinson_index_col_name.clone(), new_columns.len() - 1);
    }

    // Add Theil Index column (requires reading raw data, computed for numeric/date types)
    // Only add if --advanced flag is set
    if args.flag_advanced && !column_exists("theil_index") {
        new_columns.push("theil_index".to_string());
        new_column_indices.insert("theil_index".to_string(), new_columns.len() - 1);
    }

    // Add Mean Absolute Deviation from mean (requires reading raw data)
    // Only add if --advanced flag is set
    if args.flag_advanced && !column_exists("mean_ad") {
        new_columns.push("mean_ad".to_string());
        new_column_indices.insert("mean_ad".to_string(), new_columns.len() - 1);
    }

    // Add Shannon Entropy column (requires reading raw data, computed for all field types)
    // Only add if --advanced flag is set
    if args.flag_advanced && !column_exists("shannon_entropy") {
        new_columns.push("shannon_entropy".to_string());
        new_column_indices.insert("shannon_entropy".to_string(), new_columns.len() - 1);
    }

    if new_column_indices.contains_key("shannon_entropy")
        && cardinality_idx.is_some()
        && !column_exists("normalized_entropy")
    {
        new_columns.push("normalized_entropy".to_string());
        new_column_indices.insert("normalized_entropy".to_string(), new_columns.len() - 1);
    }

    // Simpson's Diversity Index: 1 - Σ(p_i²)
    // Computed alongside entropy from frequency data, works for all field types
    if new_column_indices.contains_key("shannon_entropy")
        && !column_exists("simpsons_diversity_index")
    {
        new_columns.push("simpsons_diversity_index".to_string());
        new_column_indices.insert(
            "simpsons_diversity_index".to_string(),
            new_columns.len() - 1,
        );
    }

    // Add XSD type column (computed for all field types based on type and min/max)
    if !column_exists("xsd_type") {
        new_columns.push("xsd_type".to_string());
        new_column_indices.insert("xsd_type".to_string(), new_columns.len() - 1);
    }

    // Add outlier count columns if all fences are available
    // Only add if at least one outlier column doesn't exist (to avoid partial duplicates)
    if lower_outer_fence_idx.is_some()
        && lower_inner_fence_idx.is_some()
        && upper_inner_fence_idx.is_some()
        && upper_outer_fence_idx.is_some()
        && !column_exists("outliers_extreme_lower_cnt")
    {
        // Count columns (with _cnt suffix)
        new_columns.push("outliers_extreme_lower_cnt".to_string());
        new_column_indices.insert(
            "outliers_extreme_lower_cnt".to_string(),
            new_columns.len() - 1,
        );
        new_columns.push("outliers_mild_lower_cnt".to_string());
        new_column_indices.insert("outliers_mild_lower_cnt".to_string(), new_columns.len() - 1);
        new_columns.push("outliers_normal_cnt".to_string());
        new_column_indices.insert("outliers_normal_cnt".to_string(), new_columns.len() - 1);
        new_columns.push("outliers_mild_upper_cnt".to_string());
        new_column_indices.insert("outliers_mild_upper_cnt".to_string(), new_columns.len() - 1);
        new_columns.push("outliers_extreme_upper_cnt".to_string());
        new_column_indices.insert(
            "outliers_extreme_upper_cnt".to_string(),
            new_columns.len() - 1,
        );
        new_columns.push("outliers_total_cnt".to_string());
        new_column_indices.insert("outliers_total_cnt".to_string(), new_columns.len() - 1);
        // Additional outlier statistics computed during outlier scanning
        new_columns.push("outliers_mean".to_string());
        new_column_indices.insert("outliers_mean".to_string(), new_columns.len() - 1);
        new_columns.push("non_outliers_mean".to_string());
        new_column_indices.insert("non_outliers_mean".to_string(), new_columns.len() - 1);
        new_columns.push("outliers_to_normal_mean_ratio".to_string());
        new_column_indices.insert(
            "outliers_to_normal_mean_ratio".to_string(),
            new_columns.len() - 1,
        );
        new_columns.push("outliers_min".to_string());
        new_column_indices.insert("outliers_min".to_string(), new_columns.len() - 1);
        new_columns.push("outliers_max".to_string());
        new_column_indices.insert("outliers_max".to_string(), new_columns.len() - 1);
        new_columns.push("outliers_range".to_string());
        new_column_indices.insert("outliers_range".to_string(), new_columns.len() - 1);
        // Additional outlier statistics: variance/stddev
        new_columns.push("outliers_stddev".to_string());
        new_column_indices.insert("outliers_stddev".to_string(), new_columns.len() - 1);
        new_columns.push("outliers_variance".to_string());
        new_column_indices.insert("outliers_variance".to_string(), new_columns.len() - 1);
        new_columns.push("non_outliers_stddev".to_string());
        new_column_indices.insert("non_outliers_stddev".to_string(), new_columns.len() - 1);
        new_columns.push("non_outliers_variance".to_string());
        new_column_indices.insert("non_outliers_variance".to_string(), new_columns.len() - 1);
        // Coefficient of variation
        new_columns.push("outliers_cv".to_string());
        new_column_indices.insert("outliers_cv".to_string(), new_columns.len() - 1);
        new_columns.push("non_outliers_cv".to_string());
        new_column_indices.insert("non_outliers_cv".to_string(), new_columns.len() - 1);
        // Outlier percentage
        new_columns.push("outliers_percentage".to_string());
        new_column_indices.insert("outliers_percentage".to_string(), new_columns.len() - 1);
        // Outlier impact
        new_columns.push("outlier_impact".to_string());
        new_column_indices.insert("outlier_impact".to_string(), new_columns.len() - 1);
        new_columns.push("outlier_impact_ratio".to_string());
        new_column_indices.insert("outlier_impact_ratio".to_string(), new_columns.len() - 1);
        // Outlier-to-normal spread ratio
        new_columns.push("outliers_normal_stddev_ratio".to_string());
        new_column_indices.insert(
            "outliers_normal_stddev_ratio".to_string(),
            new_columns.len() - 1,
        );
        // Z-scores of outlier boundaries
        new_columns.push("lower_outer_fence_zscore".to_string());
        new_column_indices.insert(
            "lower_outer_fence_zscore".to_string(),
            new_columns.len() - 1,
        );
        new_columns.push("upper_outer_fence_zscore".to_string());
        new_column_indices.insert(
            "upper_outer_fence_zscore".to_string(),
            new_columns.len() - 1,
        );
    }

    // Add winsorized and trimmed mean columns
    // Check if we can add winsorized/trimmed means
    // Need either Q1/Q3 (default) or percentiles (with --use-percentiles)
    let can_add_winsorized_trimmed = if args.flag_use_percentiles {
        percentiles_idx.is_some()
    } else {
        q1_idx.is_some() && q3_idx.is_some()
    };

    // Determine column names for winsorized/trimmed means
    let (winsorized_col_name, trimmed_col_name) = if args.flag_use_percentiles {
        if let (Some(lower_pct), Some(_upper_pct)) = (lower_percentile, upper_percentile) {
            let pct_str = format!("{}pct", fmt_pct(lower_pct));
            (
                format!("winsorized_mean_{pct_str}"),
                format!("trimmed_mean_{pct_str}"),
            )
        } else {
            (
                "winsorized_mean_5pct".to_string(),
                "trimmed_mean_5pct".to_string(),
            )
        }
    } else {
        (
            "winsorized_mean_25pct".to_string(),
            "trimmed_mean_25pct".to_string(),
        )
    };

    if can_add_winsorized_trimmed && !column_exists(winsorized_col_name.as_str()) {
        new_columns.push(winsorized_col_name.clone());
        new_column_indices.insert(winsorized_col_name.clone(), new_columns.len() - 1);
        new_columns.push(trimmed_col_name.clone());
        new_column_indices.insert(trimmed_col_name.clone(), new_columns.len() - 1);
        // Add trimmed/winsorized variance and stddev columns
        let trimmed_stddev_name = trimmed_col_name.replace("mean", "stddev");
        let trimmed_variance_name = trimmed_col_name.replace("mean", "variance");
        let winsorized_stddev_name = winsorized_col_name.replace("mean", "stddev");
        let winsorized_variance_name = winsorized_col_name.replace("mean", "variance");
        new_columns.push(trimmed_stddev_name.clone());
        new_column_indices.insert(trimmed_stddev_name, new_columns.len() - 1);
        new_columns.push(trimmed_variance_name.clone());
        new_column_indices.insert(trimmed_variance_name, new_columns.len() - 1);
        new_columns.push(winsorized_stddev_name.clone());
        new_column_indices.insert(winsorized_stddev_name, new_columns.len() - 1);
        new_columns.push(winsorized_variance_name.clone());
        new_column_indices.insert(winsorized_variance_name, new_columns.len() - 1);
        // Add trimmed/winsorized coefficient of variation
        let trimmed_cv_name = trimmed_col_name.replace("mean", "cv");
        let winsorized_cv_name = winsorized_col_name.replace("mean", "cv");
        new_columns.push(trimmed_cv_name.clone());
        new_column_indices.insert(trimmed_cv_name, new_columns.len() - 1);
        new_columns.push(winsorized_cv_name.clone());
        new_column_indices.insert(winsorized_cv_name, new_columns.len() - 1);
        // Add robust spread ratios (replace "mean" with empty string and clean up double
        // underscores)
        let trimmed_base = trimmed_col_name.replace("mean", "").replace("__", "_");
        let winsorized_base = winsorized_col_name.replace("mean", "").replace("__", "_");
        let trimmed_stddev_ratio_name =
            format!("{}_stddev_ratio", trimmed_base.trim_end_matches('_'));
        let winsorized_stddev_ratio_name =
            format!("{}_stddev_ratio", winsorized_base.trim_end_matches('_'));
        new_columns.push(trimmed_stddev_ratio_name.clone());
        new_column_indices.insert(trimmed_stddev_ratio_name, new_columns.len() - 1);
        new_columns.push(winsorized_stddev_ratio_name.clone());
        new_column_indices.insert(winsorized_stddev_ratio_name, new_columns.len() - 1);
        // Add trimmed/winsorized range
        let trimmed_range_name = trimmed_col_name.replace("mean", "range");
        let winsorized_range_name = winsorized_col_name.replace("mean", "range");
        new_columns.push(trimmed_range_name.clone());
        new_column_indices.insert(trimmed_range_name, new_columns.len() - 1);
        new_columns.push(winsorized_range_name.clone());
        new_column_indices.insert(winsorized_range_name, new_columns.len() - 1);
    }

    if new_columns.is_empty() {
        // Check if any moarstats columns already exist to determine the reason
        let moarstats_columns = [
            "pearson_skewness",
            "range_stddev_ratio",
            "quartile_coefficient_dispersion",
            "mode_zscore",
            "relative_standard_error",
            "min_zscore",
            "max_zscore",
            "median_mean_ratio",
            "iqr_range_ratio",
            "mad_stddev_ratio",
            "trimean",
            "midhinge",
            "robust_cv",
            "kurtosis",
            "bimodality_coefficient",
            "jarque_bera",
            "jarque_bera_pvalue",
            "gini_coefficient",
            // atkinson_index headers are parameterized (e.g. atkinson_index_(1)) and
            // are matched separately via the starts_with("atkinson_index_") check below.
            "theil_index",
            "mean_ad",
            "shannon_entropy",
            "normalized_entropy",
            "simpsons_diversity_index",
            "xsd_type",
            "outliers_extreme_lower_cnt",
        ];

        let any_exist = moarstats_columns.iter().any(|col| column_exists(col))
            || headers.iter().any(|h| h.starts_with("atkinson_index_"));

        if any_exist {
            wwarn!(
                "Warning: No additional stats can be computed. All available additional \
                 statistics have already been added to this stats CSV file."
            );
        } else {
            wwarn!(
                "Warning: No additional stats can be computed with the available base statistics."
            );
            wwarn!(
                "Consider running stats with --everything, or including --quartiles --median \
                 --mode in your --stats-options."
            );
        }
        // If bivariate statistics are not requested, we can return early
        if !args.flag_bivariate {
            return Ok(());
        }
    }

    // Read all records
    let mut records = Vec::new();
    for result in rdr.records() {
        let record = result?;
        records.push(record);
    }

    // Collect fields that need outlier counting and/or winsorized/trimmed means
    let mut fields_to_count: HashMap<String, OutlierFieldInfo> = HashMap::new();
    let needs_outlier_counting = new_column_indices.contains_key("outliers_extreme_lower_cnt");
    let needs_winsorized_trimmed = new_column_indices.contains_key(winsorized_col_name.as_str())
        || new_column_indices.contains_key(trimmed_col_name.as_str());

    // Collect fields that need Kurtosis, Gini & Atkinson Index computation
    // (with their precalculated stats)
    let needs_kga = new_column_indices.contains_key("kurtosis")
        || new_column_indices.contains_key("gini_coefficient")
        || new_column_indices.contains_key(&atkinson_index_col_name)
        || new_column_indices.contains_key("theil_index")
        || new_column_indices.contains_key("mean_ad");

    // First pass: collect field information from stats records
    if needs_outlier_counting || needs_winsorized_trimmed {
        for record in &records {
            let field_name = field_idx.and_then(|idx| record.get(idx)).unwrap_or("");
            let field_type_str = record.get(type_idx).unwrap_or("");

            // Convert string to enum for efficient comparisons
            let Some(field_type) = FieldType::from_str(field_type_str) else {
                continue;
            };

            if field_name.is_empty() || !field_type.is_numeric_or_date_type() {
                continue;
            }

            // Parse fence values (needed for outlier counting)
            let lower_outer_fence = lower_outer_fence_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let lower_inner_fence = lower_inner_fence_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let upper_inner_fence = upper_inner_fence_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let upper_outer_fence = upper_outer_fence_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);

            // Parse threshold values for winsorization/trimming
            let (lower_threshold, upper_threshold) = if args.flag_use_percentiles {
                // Use percentiles
                if let (Some(percentiles_idx_val), Some(lower_pct), Some(upper_pct)) =
                    (percentiles_idx, lower_percentile, upper_percentile)
                {
                    let percentiles_str = record.get(percentiles_idx_val).unwrap_or("");
                    let lower_pct_str = fmt_pct(lower_pct);
                    let upper_pct_str = fmt_pct(upper_pct);

                    let lower_val = parse_percentile_value(
                        percentiles_str,
                        &lower_pct_str,
                        field_type,
                        &stats_separator,
                        prefer_dmy,
                    );
                    let upper_val = parse_percentile_value(
                        percentiles_str,
                        &upper_pct_str,
                        field_type,
                        &stats_separator,
                        prefer_dmy,
                    );
                    (lower_val, upper_val)
                } else {
                    (None, None)
                }
            } else {
                // Use Q1/Q3
                let q1_val = if field_type.is_date_or_datetime() {
                    q1_idx
                        .and_then(|idx| record.get(idx))
                        .and_then(|s| parse_date_to_days(s, prefer_dmy))
                } else {
                    q1_idx
                        .and_then(|idx| record.get(idx))
                        .and_then(parse_float_opt)
                };
                let q3_val = if field_type.is_date_or_datetime() {
                    q3_idx
                        .and_then(|idx| record.get(idx))
                        .and_then(|s| parse_date_to_days(s, prefer_dmy))
                } else {
                    q3_idx
                        .and_then(|idx| record.get(idx))
                        .and_then(parse_float_opt)
                };
                (q1_val, q3_val)
            };

            // Determine if we should include this field
            let include_for_outliers = needs_outlier_counting
                && lower_outer_fence.is_some()
                && lower_inner_fence.is_some()
                && upper_inner_fence.is_some()
                && upper_outer_fence.is_some();

            let include_for_winsorized_trimmed =
                needs_winsorized_trimmed && lower_threshold.is_some() && upper_threshold.is_some();

            if include_for_outliers || include_for_winsorized_trimmed {
                // Use default values for fences if not needed
                let lower_outer = lower_outer_fence.unwrap_or(0.0);
                let lower_inner = lower_inner_fence.unwrap_or(0.0);
                let upper_inner = upper_inner_fence.unwrap_or(0.0);
                let upper_outer = upper_outer_fence.unwrap_or(0.0);
                let lower_thresh = lower_threshold.unwrap_or(0.0);
                let upper_thresh = upper_threshold.unwrap_or(0.0);

                // We'll find the column index when we read the CSV
                fields_to_count.insert(
                    field_name.to_string(),
                    OutlierFieldInfo {
                        col_idx: 0, // Will be set when we read CSV headers
                        field_type, // Store enum directly
                        lower_outer,
                        lower_inner,
                        upper_inner,
                        upper_outer,
                        lower_threshold: lower_thresh,
                        upper_threshold: upper_thresh,
                    },
                );
            }
        }
    }

    // Collect fields for Kurtosis, Gini & Atkinson Index computation with their precalculated stats
    let mut fields_for_kga: HashMap<String, KGAFieldInfo> = HashMap::new();
    if needs_kga {
        for record in &records {
            let field_name = field_idx.and_then(|idx| record.get(idx)).unwrap_or("");
            let field_type_str = record.get(type_idx).unwrap_or("");

            // Convert string to enum for efficient comparisons
            let Some(field_type) = FieldType::from_str(field_type_str) else {
                continue;
            };

            if field_name.is_empty() || !field_type.is_numeric_or_date_type() {
                continue;
            }

            // Parse precalculated stats
            let mean_val = mean_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let stddev_val = stddev_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let variance_val = stddev_val.map(|s| s * s); // variance = stddev^2
            let sum_val = sum_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);

            // We'll find the column index when we read the CSV
            fields_for_kga.insert(
                field_name.to_string(),
                KGAFieldInfo {
                    col_idx: 0, // Will be set when we read CSV headers
                    field_type,
                    mean: mean_val,
                    variance: variance_val,
                    sum: sum_val,
                },
            );
        }
    }

    // Count outliers for all fields in a single pass through the original CSV
    let outlier_counts = if fields_to_count.is_empty() {
        HashMap::new()
    } else {
        // Get headers to map field names to column indices
        let mut csv_rdr = ReaderBuilder::new()
            .has_headers(true)
            .from_path(actual_input_path)?;
        let csv_headers = csv_rdr.headers()?.clone();

        // Update column indices in fields_to_count and remove fields not found in CSV
        fields_to_count.retain(|field_name, field_info| {
            if let Some(col_idx) = csv_headers.iter().position(|h| h == field_name) {
                field_info.col_idx = col_idx;
                true
            } else {
                false
            }
        });

        // Count outliers (will use parallel processing if index exists).
        // Pass fields_to_count by value so the parallel path can move it
        // directly into an Arc without an extra deep-clone.
        count_all_outliers(fields_to_count, actual_input_path, args.flag_jobs)?
    };

    // Compute kurtosis, Gini coefficient & Atkinson Index for all fields
    let kga_stats = if fields_for_kga.is_empty() {
        HashMap::new()
    } else {
        // Get headers to map field names to column indices
        let mut csv_rdr = ReaderBuilder::new()
            .has_headers(true)
            .from_path(actual_input_path)?;
        let csv_headers = csv_rdr.headers()?.clone();

        // Update column indices in fields_for_kga and remove fields not found in CSV
        fields_for_kga.retain(|field_name, field_info| {
            if let Some(col_idx) = csv_headers.iter().position(|h| h == field_name) {
                field_info.col_idx = col_idx;
                true
            } else {
                false
            }
        });

        // Compute Kurtosis, Gini & Atkinson Index (will use sequential processing for correctness)
        compute_all_kga(&fields_for_kga, actual_input_path, args.flag_epsilon)?
    };

    // Compute Shannon Entropy for all fields
    let entropy_stats = if new_column_indices.contains_key("shannon_entropy") {
        compute_all_entropy(actual_input_path)?
    } else {
        HashMap::new()
    };

    let mut stats_config = BivariateStatsConfig::default();
    // Compute bivariate statistics if requested
    // Store field_names for output conversion (indices -> names)
    let mut bivariate_field_names: Option<Vec<String>> = None;
    let bivariate_stats = if args.flag_bivariate {
        // Validate bivariate stats config early
        stats_config = BivariateStatsConfig::from_flag(&args.flag_bivariate_stats)?;

        // Get record count to check for all-unique fields (cardinality == rowcount)
        let record_count: Option<u64> = {
            let actual_input_path_str = actual_input_path
                .to_str()
                .ok_or_else(|| CliError::Other("Invalid input path".to_string()))?
                .to_string();
            let rconfig = Config::new(Some(&actual_input_path_str));
            if let Ok(Some(idx)) = rconfig.indexed() {
                Some(idx.count())
            } else if !rconfig.is_stdin() {
                // Fall back to counting rows if no index
                util::count_rows(&rconfig).ok()
            } else {
                None // Can't get count from stdin
            }
        };

        // Get CSV headers to map field names to column indices
        let mut csv_rdr = ReaderBuilder::new()
            .has_headers(true)
            .from_path(actual_input_path)?;
        let csv_headers = csv_rdr.headers()?.clone();

        // Store field names for index-to-name lookups (used for output and frequency cache)
        let field_names: Vec<String> = csv_headers
            .iter()
            .map(std::string::ToString::to_string)
            .collect();
        bivariate_field_names = Some(field_names.clone());

        // Collect all field pairs for bivariate computation using column indices as keys
        // Using u16 for keys (2 bytes) instead of usize (8 bytes) for better memory efficiency
        // u16 supports up to 65,535 columns, which is more than sufficient for any CSV
        let mut field_pairs: HashMap<(u16, u16), (BivariateFieldInfo, BivariateFieldInfo)> =
            HashMap::new();

        // Collect all numeric/date/string field names from stats CSV
        let stats_field_names: Vec<String> = records
            .iter()
            .filter_map(|r| {
                field_idx
                    .and_then(|idx| r.get(idx))
                    .map(std::string::ToString::to_string)
            })
            .collect();

        for (i, field1_name) in stats_field_names.iter().enumerate() {
            let field1_type_str = records.get(i).and_then(|r| r.get(type_idx)).unwrap_or("");
            let Some(field1_type) = FieldType::from_str(field1_type_str) else {
                continue;
            };

            // Get column index for field1
            let Some(field1_col_idx) = csv_headers.iter().position(|h| h == field1_name) else {
                continue;
            };

            // Extract pre-computed statistics for field1 from stats CSV
            let field1_record = records.get(i);
            let field1_stddev = field1_record
                .and_then(|r| stddev_idx.and_then(|idx| r.get(idx)))
                .and_then(parse_float_opt);
            let field1_variance = field1_record
                .and_then(|r| variance_idx.and_then(|idx| r.get(idx)))
                .and_then(parse_float_opt);
            let field1_cardinality = field1_record
                .and_then(|r| cardinality_idx.and_then(|idx| r.get(idx)))
                .and_then(|s| s.parse::<u64>().ok());

            // Compare with all other fields
            for (j, field2_name) in stats_field_names.iter().enumerate().skip(i + 1) {
                let field2_type_str = records.get(j).and_then(|r| r.get(type_idx)).unwrap_or("");
                let Some(field2_type) = FieldType::from_str(field2_type_str) else {
                    continue;
                };

                // Get column index for field2
                let Some(field2_col_idx) = csv_headers.iter().position(|h| h == field2_name) else {
                    continue;
                };

                // Extract pre-computed statistics for field2 from stats CSV
                let field2_record = records.get(j);
                let field2_stddev = field2_record
                    .and_then(|r| stddev_idx.and_then(|idx| r.get(idx)))
                    .and_then(parse_float_opt);
                let field2_variance = field2_record
                    .and_then(|r| variance_idx.and_then(|idx| r.get(idx)))
                    .and_then(parse_float_opt);
                let field2_cardinality = field2_record
                    .and_then(|r| cardinality_idx.and_then(|idx| r.get(idx)))
                    .and_then(|s| s.parse::<u64>().ok());

                // Filter invalid pairs: skip constant fields (zero variance)
                if let (Some(stddev1), Some(stddev2)) = (field1_stddev, field2_stddev) {
                    if stddev1.abs() < f64::EPSILON || stddev2.abs() < f64::EPSILON {
                        continue; // Skip pairs with constant fields (correlation undefined)
                    }
                } else if let (Some(var1), Some(var2)) = (field1_variance, field2_variance)
                    && (var1.abs() < f64::EPSILON || var2.abs() < f64::EPSILON)
                {
                    continue; // Skip pairs with constant fields (correlation undefined)
                }

                // Filter invalid pairs: skip both-constant pairs (cardinality = 1 for both)
                if let (Some(card1), Some(card2)) = (field1_cardinality, field2_cardinality)
                    && card1 == 1
                    && card2 == 1
                {
                    continue; // Both constant, no meaningful correlation
                }

                // Filter invalid pairs: skip fields with all unique values (cardinality ==
                // rowcount)
                if let Some(rowcount) = record_count
                    && (field1_cardinality.is_some_and(|c| c == rowcount)
                        || field2_cardinality.is_some_and(|c| c == rowcount))
                {
                    continue; // All values are unique, correlations are not meaningful
                }

                // Include pairs where at least one field is numeric/date/string
                // (for mutual information, we want all types)
                if field1_type.is_numeric_or_date_type()
                    || field2_type.is_numeric_or_date_type()
                    || field1_type == FieldType::TString
                    || field2_type == FieldType::TString
                {
                    // Use column indices as keys (cast to u16 for memory efficiency)
                    // col_idx is usize but we store as u16 in the HashMap key
                    field_pairs.insert(
                        (field1_col_idx as u16, field2_col_idx as u16),
                        (
                            BivariateFieldInfo {
                                col_idx:     field1_col_idx,
                                field_type:  field1_type,
                                stddev:      field1_stddev,
                                variance:    field1_variance,
                                cardinality: field1_cardinality,
                            },
                            BivariateFieldInfo {
                                col_idx:     field2_col_idx,
                                field_type:  field2_type,
                                stddev:      field2_stddev,
                                variance:    field2_variance,
                                cardinality: field2_cardinality,
                            },
                        ),
                    );
                }
            }
        }

        if field_pairs.is_empty() {
            HashMap::new()
        } else {
            // Setup progress bar if requested and not reading from stdin
            let actual_input_path_str = actual_input_path
                .to_str()
                .ok_or_else(|| CliError::Other("Invalid input path".to_string()))?
                .to_string();
            let rconfig_bivariate = Config::new(Some(&actual_input_path_str));
            let show_progress = (args.flag_progressbar || util::get_envvar_flag("QSV_PROGRESSBAR"))
                && !rconfig_bivariate.is_stdin();
            let progress = if show_progress {
                Some(ProgressBar::with_draw_target(
                    Some(0),
                    ProgressDrawTarget::stderr_with_hz(5),
                ))
            } else {
                None
            };

            // Get cardinality threshold (default: 1,000,000)
            let cardinality_threshold = args.flag_cardinality_threshold.or(Some(1_000_000));

            // Log which stats are being computed
            let stats_list: Vec<&str> = [
                stats_config.pearson.then_some("pearson"),
                stats_config.spearman.then_some("spearman"),
                stats_config.kendall.then_some("kendall"),
                stats_config.covariance.then_some("covariance"),
                stats_config.mi.then_some("mi"),
                stats_config.nmi.then_some("nmi"),
            ]
            .into_iter()
            .flatten()
            .collect();
            winfo!(
                "Computing bivariate statistics: {}...",
                stats_list.join(", ")
            );

            let result = compute_all_bivariatestats(
                field_pairs,
                &field_names,
                actual_input_path,
                progress.as_ref(),
                cardinality_threshold,
                stats_config,
                args.flag_jobs,
            );

            // Clean up progress bar if it was created
            if let Some(pb) = progress {
                pb.finish_and_clear();
            }

            result?
        }
    } else {
        HashMap::new()
    };

    // Write bivariate statistics CSV if computed
    // Always use the original input path for naming, even if we joined datasets
    if args.flag_bivariate && !bivariate_stats.is_empty() {
        let is_joined = temp_joined_path.is_some();
        let bivariate_csv_path = get_bivariate_csv_path(input_path, is_joined)?;
        let mut bivariate_wtr = WriterBuilder::new()
            .has_headers(true)
            .from_path(&bivariate_csv_path)?;

        // Build headers dynamically based on requested stats
        let mut headers = vec!["field1", "field2"];
        if stats_config.pearson {
            headers.push("pearson_correlation");
        }
        if stats_config.spearman {
            headers.push("spearman_correlation");
        }
        if stats_config.kendall {
            headers.push("kendall_tau");
        }
        if stats_config.covariance {
            headers.push("covariance_sample");
            headers.push("covariance_population");
        }
        if stats_config.mi {
            headers.push("mutual_information");
        }
        if stats_config.nmi {
            headers.push("normalized_mutual_information");
        }
        headers.push("n_pairs");

        // Write headers
        bivariate_wtr.write_record(&headers)?;

        // Write bivariate statistics
        // Convert indices to names for output
        let field_names_for_output = bivariate_field_names.as_ref().ok_or_else(|| {
            CliError::Other("Field names not available for bivariate output".to_string())
        })?;

        let mut sorted_pairs: Vec<_> = bivariate_stats.keys().collect();
        sorted_pairs.sort();

        for (idx1, idx2) in sorted_pairs {
            if let Some(stats) = bivariate_stats.get(&(*idx1, *idx2)) {
                // Convert indices to field names for output (u16 -> usize for indexing)
                let field1_name = field_names_for_output
                    .get(*idx1 as usize)
                    .map_or("?", |s| s.as_str());
                let field2_name = field_names_for_output
                    .get(*idx2 as usize)
                    .map_or("?", |s| s.as_str());

                // Build record dynamically based on requested stats
                let mut record: Vec<String> =
                    vec![field1_name.to_string(), field2_name.to_string()];
                if stats_config.pearson {
                    record.push(
                        stats
                            .pearson
                            .map_or(String::new(), |v| util::round_num(v, args.flag_round)),
                    );
                }
                if stats_config.spearman {
                    record.push(
                        stats
                            .spearman
                            .map_or(String::new(), |v| util::round_num(v, args.flag_round)),
                    );
                }
                if stats_config.kendall {
                    record.push(
                        stats
                            .kendall
                            .map_or(String::new(), |v| util::round_num(v, args.flag_round)),
                    );
                }
                if stats_config.covariance {
                    record.push(
                        stats
                            .covariance_sample
                            .map_or(String::new(), |v| util::round_num(v, args.flag_round)),
                    );
                    record.push(
                        stats
                            .covariance_population
                            .map_or(String::new(), |v| util::round_num(v, args.flag_round)),
                    );
                }
                if stats_config.mi {
                    record.push(
                        stats
                            .mutual_information
                            .map_or(String::new(), |v| util::round_num(v, args.flag_round)),
                    );
                }
                if stats_config.nmi {
                    record.push(
                        stats
                            .normalized_mutual_information
                            .map_or(String::new(), |v| util::round_num(v, args.flag_round)),
                    );
                }
                record.push(stats.n_pairs.to_string());

                bivariate_wtr.write_record(&record)?;
            }
        }

        bivariate_wtr.flush()?;
        wwarn!(
            "Wrote bivariate statistics to {}",
            bivariate_csv_path.display()
        );
    }

    // Prepare output
    let output_path: &Path = args.flag_output.as_ref().map_or(&stats_csv_path, Path::new);
    let mut wtr = WriterBuilder::new()
        .has_headers(true)
        .from_path(output_path)?;

    // Write headers with new columns appended
    let mut header_record = headers;
    for col in &new_columns {
        header_record.push_field(col.as_str());
    }
    wtr.write_record(&header_record)?;

    // Pre-compute derived column names for winsorized/trimmed stats
    // to avoid repeated String::replace() allocations in the per-record loop
    let winsorized_stddev_name = winsorized_col_name.replace("mean", "stddev");
    let winsorized_variance_name = winsorized_col_name.replace("mean", "variance");
    let winsorized_cv_name = winsorized_col_name.replace("mean", "cv");
    let winsorized_base = winsorized_col_name.replace("mean", "").replace("__", "_");
    let winsorized_stddev_ratio_name =
        format!("{}_stddev_ratio", winsorized_base.trim_end_matches('_'));
    let winsorized_range_name = winsorized_col_name.replace("mean", "range");
    let trimmed_stddev_name = trimmed_col_name.replace("mean", "stddev");
    let trimmed_variance_name = trimmed_col_name.replace("mean", "variance");
    let trimmed_cv_name = trimmed_col_name.replace("mean", "cv");
    let trimmed_base = trimmed_col_name.replace("mean", "").replace("__", "_");
    let trimmed_stddev_ratio_name = format!("{}_stddev_ratio", trimmed_base.trim_end_matches('_'));
    let trimmed_range_name = trimmed_col_name.replace("mean", "range");

    // Pre-allocate new_values outside the loop and reuse across iterations
    let new_columns_len = new_columns.len();
    let mut new_values: Vec<String> = vec![String::new(); new_columns_len];

    // Process each record
    #[allow(clippy::cast_precision_loss)]
    for record in &records {
        // Get field name and type (skip dataset stats rows that might not have proper type)
        let field_name = field_idx.and_then(|idx| record.get(idx)).unwrap_or("");
        let field_type_str = record.get(type_idx).unwrap_or("");

        // Convert string to enum for efficient comparisons
        let field_type_opt = FieldType::from_str(field_type_str);

        // Clear new_values for reuse (reset each string to empty without deallocating)
        for v in &mut new_values {
            v.clear();
        }

        // Compute XSD type for all field types (needs type, min, max)
        if new_column_indices.contains_key("xsd_type") {
            // Extract min and max string values (needed for comprehensive mode and as fallback)
            let min_str = min_idx.and_then(|idx| {
                let s = record.get(idx)?;
                if s.is_empty() { None } else { Some(s) }
            });
            let max_str = max_idx.and_then(|idx| {
                let s = record.get(idx)?;
                if s.is_empty() { None } else { Some(s) }
            });

            // Extract percentile values for thorough mode
            let (percentile_values, actual_scan_mode) = if scan_mode == "thorough" {
                if let Some(idx) = percentiles_idx {
                    if let Some(percentiles_str) = record.get(idx) {
                        if percentiles_str.is_empty() {
                            // Empty percentile string, fall back to quick
                            (None, "quick")
                        } else {
                            let values = parse_all_percentile_string_values(
                                percentiles_str,
                                &stats_separator,
                            );
                            if values.is_empty() {
                                // Empty percentile values, fall back to quick
                                (None, "quick")
                            } else {
                                (Some(values), "thorough")
                            }
                        }
                    } else {
                        // No percentile string, fall back to quick
                        (None, "quick")
                    }
                } else {
                    // No percentiles column, fall back to quick
                    (None, "quick")
                }
            } else {
                (None, scan_mode)
            };

            // Parse min and max values - they may be strings (for dates) or numbers (for
            // integers/floats)
            let min_val = if let Some(min_idx_val) = min_idx {
                record.get(min_idx_val).and_then(|s| {
                    if s.is_empty() {
                        None
                    } else if field_type_opt.is_some_and(FieldType::is_date_or_datetime) {
                        parse_date_to_days(s, prefer_dmy)
                    } else {
                        parse_float_opt(s)
                    }
                })
            } else {
                None
            };

            let max_val = if let Some(max_idx_val) = max_idx {
                record.get(max_idx_val).and_then(|s| {
                    if s.is_empty() {
                        None
                    } else if field_type_opt.is_some_and(FieldType::is_date_or_datetime) {
                        parse_date_to_days(s, prefer_dmy)
                    } else {
                        parse_float_opt(s)
                    }
                })
            } else {
                None
            };

            // Infer XSD type (pass all parameters including scan_mode and percentile_values)
            // Use actual_scan_mode which may have fallen back to quick if percentiles unavailable
            let xsd_type = infer_xsd_type(
                field_type_str,
                min_val,
                max_val,
                field_type_opt,
                actual_scan_mode,
                min_str,
                max_str,
                percentile_values.as_deref(),
            );
            if let Some(idx) = new_column_indices.get("xsd_type") {
                new_values[*idx] = xsd_type;
            }
        }

        // Write Shannon Entropy from pre-computed results (works for all field types)
        if new_column_indices.contains_key("shannon_entropy")
            && !field_name.is_empty()
            && let Some(stats) = entropy_stats.get(field_name)
            && let Some(entropy_val) = stats.entropy
            && let Some(idx) = new_column_indices.get("shannon_entropy")
        {
            new_values[*idx] = util::round_num(entropy_val, args.flag_round);
        }

        // Write Normalized Entropy from pre-computed results (works for all field types)
        if let Some(idx) = new_column_indices.get("normalized_entropy")
            && !field_name.is_empty()
            && let Some(entropy_stats) = entropy_stats.get(field_name)
            && let Some(entropy_val) = entropy_stats.entropy
        {
            let cardinality_val = cardinality_idx
                .and_then(|idx| record.get(idx))
                .and_then(|s| s.parse::<u64>().ok());
            if let Some(val) = compute_normalized_entropy(Some(entropy_val), cardinality_val) {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }
        }

        // Write Simpson's Diversity Index from pre-computed results (works for all field types)
        if let Some(idx) = new_column_indices.get("simpsons_diversity_index")
            && !field_name.is_empty()
            && let Some(entropy_stats_val) = entropy_stats.get(field_name)
            && let Some(simpsons_val) = entropy_stats_val.simpsons_diversity
        {
            new_values[*idx] = util::round_num(simpsons_val, args.flag_round);
        }

        // Only compute other stats for numeric/date types
        let Some(field_type) = field_type_opt else {
            // For unrecognized types, write existing fields + new values directly
            for field in record {
                wtr.write_field(field)?;
            }
            for val in &new_values {
                wtr.write_field(val)?;
            }
            wtr.write_record(None::<&[u8]>)?;
            continue;
        };

        if field_type.is_numeric_or_date_type() {
            // Parse existing stats values
            let mean = mean_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let median = median_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt)
                .or_else(|| {
                    q2_median_idx
                        .and_then(|idx| record.get(idx))
                        .and_then(parse_float_opt)
                });
            let stddev = stddev_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let range = range_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let q1 = q1_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let q3 = q3_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);

            // Parse mode (may be a string, need to try parsing as float)
            // If multiple modes are separated by "|", try parsing the first one
            let mode = mode_idx.and_then(|idx| record.get(idx)).and_then(|s| {
                if s.is_empty() {
                    None
                } else {
                    // Handle multiple modes separated by "|" - try first one
                    // safety: `split` on a non-empty string always yields at least one element,
                    // so `next` will always return `Some` and `unwrap` will not panic.
                    let first_mode = s.split('|').next().unwrap().trim();
                    parse_float_opt(first_mode)
                }
            });

            // Parse additional stats
            let sem = sem_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let min = min_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let max = max_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let iqr = iqr_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);
            let mad = mad_idx
                .and_then(|idx| record.get(idx))
                .and_then(parse_float_opt);

            // Compute new stats (entropy already computed above for all field types)

            if let Some(idx) = new_column_indices.get("pearson_skewness")
                && let Some(val) = compute_pearson_skewness(mean, median, stddev)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("range_stddev_ratio")
                && let Some(val) = compute_range_stddev_ratio(range, stddev)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("quartile_coefficient_dispersion")
                && let Some(val) = compute_quartile_coefficient_dispersion(q1, q3)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("mode_zscore")
                && let Some(val) = compute_mode_zscore(mode, mean, stddev)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("relative_standard_error")
                && let Some(val) = compute_relative_standard_error(sem, mean)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("min_zscore")
                && let Some(val) = compute_zscore(min, mean, stddev)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("max_zscore")
                && let Some(val) = compute_zscore(max, mean, stddev)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("median_mean_ratio")
                && let Some(val) = compute_median_mean_ratio(median, mean)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("iqr_range_ratio")
                && let Some(val) = compute_iqr_range_ratio(iqr, range)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("mad_stddev_ratio")
                && let Some(val) = compute_mad_stddev_ratio(mad, stddev)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("trimean")
                && let Some(val) = compute_trimean(q1, median, q3)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("midhinge")
                && let Some(val) = compute_midhinge(q1, q3)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            if let Some(idx) = new_column_indices.get("robust_cv")
                && let Some(val) = compute_robust_cv(mad, median)
            {
                new_values[*idx] = util::round_num(val, args.flag_round);
            }

            // Compute Bimodality Coefficient (requires skewness and kurtosis)
            if let Some(idx) = new_column_indices.get("bimodality_coefficient")
                && !field_name.is_empty()
                && let Some(kga_stats_val) = kga_stats.get(field_name)
                && let Some(kurtosis_val) = kga_stats_val.kurtosis
            {
                let skewness = skewness_idx
                    .and_then(|idx| record.get(idx))
                    .and_then(parse_float_opt);
                if let Some(val) = compute_bimodality_coefficient(skewness, Some(kurtosis_val)) {
                    new_values[*idx] = util::round_num(val, args.flag_round);
                }
            }

            // Compute Jarque-Bera test (requires skewness and kurtosis)
            // Prefer kurtosis from KGA stats when available, otherwise fall back
            // to the kurtosis value already present in the stats CSV record.
            if new_column_indices.contains_key("jarque_bera") && !field_name.is_empty() {
                let kurtosis_from_kga = kga_stats
                    .get(field_name)
                    .and_then(|kga_stats_val| kga_stats_val.kurtosis);
                let kurtosis_from_stats = kurtosis_idx
                    .and_then(|idx| record.get(idx))
                    .and_then(parse_float_opt);
                let kurtosis_val = kurtosis_from_kga.or(kurtosis_from_stats);

                if let Some(kurtosis_val) = kurtosis_val {
                    let skewness = skewness_idx
                        .and_then(|idx| record.get(idx))
                        .and_then(parse_float_opt);
                    // Compute n from n_positive + n_negative + n_zero
                    let n_pos = n_positive_idx
                        .and_then(|idx| record.get(idx))
                        .and_then(|s| s.parse::<u64>().ok())
                        .unwrap_or(0);
                    let n_neg = n_negative_idx
                        .and_then(|idx| record.get(idx))
                        .and_then(|s| s.parse::<u64>().ok())
                        .unwrap_or(0);
                    let n_z = n_zero_idx
                        .and_then(|idx| record.get(idx))
                        .and_then(|s| s.parse::<u64>().ok())
                        .unwrap_or(0);
                    let n = n_pos + n_neg + n_z;
                    if let Some((jb, pval)) = compute_jarque_bera(skewness, Some(kurtosis_val), n) {
                        if let Some(idx) = new_column_indices.get("jarque_bera") {
                            new_values[*idx] = util::round_num(jb, args.flag_round);
                        }
                        if let Some(idx) = new_column_indices.get("jarque_bera_pvalue") {
                            new_values[*idx] = util::round_num(pval, args.flag_round);
                        }
                    }
                }
            }

            // Get outlier statistics from pre-computed results
            if new_column_indices.contains_key("outliers_extreme_lower_cnt")
                && !field_name.is_empty()
                && let Some(stats) = outlier_counts.get(field_name)
            {
                // Write counts (with _cnt suffix)
                if let Some(idx) = new_column_indices.get("outliers_extreme_lower_cnt") {
                    new_values[*idx] = stats.counts[OUTLIER_EXTREME_LOWER].to_string();
                }
                if let Some(idx) = new_column_indices.get("outliers_mild_lower_cnt") {
                    new_values[*idx] = stats.counts[OUTLIER_MILD_LOWER].to_string();
                }
                if let Some(idx) = new_column_indices.get("outliers_normal_cnt") {
                    new_values[*idx] = stats.counts[OUTLIER_NORMAL].to_string();
                }
                if let Some(idx) = new_column_indices.get("outliers_mild_upper_cnt") {
                    new_values[*idx] = stats.counts[OUTLIER_MILD_UPPER].to_string();
                }
                if let Some(idx) = new_column_indices.get("outliers_extreme_upper_cnt") {
                    new_values[*idx] = stats.counts[OUTLIER_EXTREME_UPPER].to_string();
                }
                if let Some(idx) = new_column_indices.get("outliers_total_cnt") {
                    new_values[*idx] = stats.counts[OUTLIER_TOTAL].to_string();
                }

                // Compute means
                let mean_outliers = if stats.counts[OUTLIER_TOTAL] > 0 {
                    Some(stats.sum_outliers / stats.counts[OUTLIER_TOTAL] as f64)
                } else {
                    None
                };
                let mean_normal = if stats.counts[OUTLIER_NORMAL] > 0 {
                    Some(stats.sum_normal / stats.counts[OUTLIER_NORMAL] as f64)
                } else {
                    None
                };
                let mean_all = if stats.count_all > 0 {
                    Some(stats.sum_all / stats.count_all as f64)
                } else {
                    None
                };

                // Compute outliers variance and stddev once for reuse
                let (variance_outliers, stddev_outliers) = if stats.counts[OUTLIER_TOTAL] > 1 {
                    let n = stats.counts[OUTLIER_TOTAL] as f64;
                    let variance = (stats.sum_squares_outliers
                        - (stats.sum_outliers * stats.sum_outliers / n))
                        / (n - 1.0);
                    if variance >= 0.0 {
                        (Some(variance), Some(variance.sqrt()))
                    } else {
                        (None, None)
                    }
                } else {
                    (None, None)
                };

                // Compute and write additional statistics
                if let Some(mean_outliers_val) = mean_outliers {
                    // Mean of outliers
                    if let Some(idx) = new_column_indices.get("outliers_mean") {
                        new_values[*idx] = if field_type.is_date_or_datetime() {
                            days_to_rfc3339(mean_outliers_val, field_type)
                        } else {
                            util::round_num(mean_outliers_val, args.flag_round)
                        };
                    }

                    // Variance and stddev of outliers
                    if let (Some(variance_outliers_val), Some(stddev_outliers_val)) =
                        (variance_outliers, stddev_outliers)
                    {
                        if let Some(idx) = new_column_indices.get("outliers_stddev") {
                            new_values[*idx] =
                                util::round_num(stddev_outliers_val, args.flag_round);
                        }
                        if let Some(idx) = new_column_indices.get("outliers_variance") {
                            new_values[*idx] =
                                util::round_num(variance_outliers_val, args.flag_round);
                        }
                        // Coefficient of variation for outliers
                        if mean_outliers_val.abs() > f64::EPSILON
                            && let Some(idx) = new_column_indices.get("outliers_cv")
                        {
                            let cv = stddev_outliers_val / mean_outliers_val.abs();
                            new_values[*idx] = util::round_num(cv, args.flag_round);
                        }
                    }
                }

                if let Some(mean_normal_val) = mean_normal {
                    // Mean of non-outliers
                    if let Some(idx) = new_column_indices.get("non_outliers_mean") {
                        new_values[*idx] = if field_type.is_date_or_datetime() {
                            days_to_rfc3339(mean_normal_val, field_type)
                        } else {
                            util::round_num(mean_normal_val, args.flag_round)
                        };
                    }

                    // Variance and stddev of non-outliers
                    if stats.counts[OUTLIER_NORMAL] > 1 {
                        let n = stats.counts[OUTLIER_NORMAL] as f64;
                        let variance_normal = (stats.sum_squares_normal
                            - (stats.sum_normal * stats.sum_normal / n))
                            / (n - 1.0);
                        if variance_normal >= 0.0 {
                            let stddev_normal = variance_normal.sqrt();
                            if let Some(idx) = new_column_indices.get("non_outliers_stddev") {
                                new_values[*idx] = util::round_num(stddev_normal, args.flag_round);
                            }
                            if let Some(idx) = new_column_indices.get("non_outliers_variance") {
                                new_values[*idx] =
                                    util::round_num(variance_normal, args.flag_round);
                            }
                            // Coefficient of variation for non-outliers
                            if mean_normal_val.abs() > f64::EPSILON
                                && let Some(idx) = new_column_indices.get("non_outliers_cv")
                            {
                                let cv = stddev_normal / mean_normal_val.abs();
                                new_values[*idx] = util::round_num(cv, args.flag_round);
                            }

                            // Outlier-to-normal spread ratio
                            if let Some(stddev_outliers_val) = stddev_outliers
                                && stddev_normal.abs() > f64::EPSILON
                                && let Some(idx) =
                                    new_column_indices.get("outliers_normal_stddev_ratio")
                            {
                                let ratio = stddev_outliers_val / stddev_normal;
                                new_values[*idx] = util::round_num(ratio, args.flag_round);
                            }
                        }
                    }

                    // Outlier-to-normal mean ratio
                    if let Some(mean_outliers_val) = mean_outliers
                        && let Some(idx) = new_column_indices.get("outliers_to_normal_mean_ratio")
                        && mean_normal_val.abs() > f64::EPSILON
                    {
                        let ratio = mean_outliers_val / mean_normal_val;
                        new_values[*idx] = util::round_num(ratio, args.flag_round);
                    }
                }

                // Outlier percentage
                if stats.count_all > 0
                    && let Some(idx) = new_column_indices.get("outliers_percentage")
                {
                    let percentage =
                        (stats.counts[OUTLIER_TOTAL] as f64 / stats.count_all as f64) * 100.0;
                    new_values[*idx] = util::round_num(percentage, args.flag_round);
                }

                // Outlier impact
                if let (Some(mean_all_val), Some(mean_normal_val)) = (mean_all, mean_normal) {
                    if let Some(idx) = new_column_indices.get("outlier_impact") {
                        let impact = mean_all_val - mean_normal_val;
                        new_values[*idx] = util::round_num(impact, args.flag_round);
                    }
                    if let Some(idx) = new_column_indices.get("outlier_impact_ratio")
                        && mean_normal_val.abs() > f64::EPSILON
                    {
                        let impact = mean_all_val - mean_normal_val;
                        let ratio = impact / mean_normal_val.abs();
                        new_values[*idx] = util::round_num(ratio, args.flag_round);
                    }
                }

                // Z-scores of outlier boundaries
                if let (Some(mean_val), Some(stddev_val)) = (mean, stddev)
                    && stddev_val.abs() > f64::EPSILON
                {
                    if let (Some(lower_outer), Some(idx)) = (
                        lower_outer_fence_idx
                            .and_then(|idx| record.get(idx))
                            .and_then(parse_float_opt),
                        new_column_indices.get("lower_outer_fence_zscore"),
                    ) {
                        let zscore = (lower_outer - mean_val) / stddev_val;
                        new_values[*idx] = util::round_num(zscore, args.flag_round);
                    }
                    if let (Some(upper_outer), Some(idx)) = (
                        upper_outer_fence_idx
                            .and_then(|idx| record.get(idx))
                            .and_then(parse_float_opt),
                        new_column_indices.get("upper_outer_fence_zscore"),
                    ) {
                        let zscore = (upper_outer - mean_val) / stddev_val;
                        new_values[*idx] = util::round_num(zscore, args.flag_round);
                    }
                }

                // Min/Max/Range of outliers
                if let Some(min_outliers) = stats.min_outliers
                    && let Some(idx) = new_column_indices.get("outliers_min")
                {
                    new_values[*idx] = if field_type.is_date_or_datetime() {
                        days_to_rfc3339(min_outliers, field_type)
                    } else {
                        util::round_num(min_outliers, args.flag_round)
                    };
                }
                if let Some(max_outliers) = stats.max_outliers {
                    if let Some(idx) = new_column_indices.get("outliers_max") {
                        new_values[*idx] = if field_type.is_date_or_datetime() {
                            days_to_rfc3339(max_outliers, field_type)
                        } else {
                            util::round_num(max_outliers, args.flag_round)
                        };
                    }
                    // Range of outliers
                    if let Some(min_outliers) = stats.min_outliers
                        && let Some(idx) = new_column_indices.get("outliers_range")
                    {
                        let range = max_outliers - min_outliers;
                        new_values[*idx] = util::round_num(range, args.flag_round);
                    }
                }
            }

            // Write winsorized and trimmed means and related statistics
            if (new_column_indices.contains_key(winsorized_col_name.as_str())
                || new_column_indices.contains_key(trimmed_col_name.as_str()))
                && !field_name.is_empty()
                && let Some(stats) = outlier_counts.get(field_name)
            {
                // Compute means
                let winsorized_mean = if stats.winsorized_count > 0 {
                    Some(stats.winsorized_sum / stats.winsorized_count as f64)
                } else {
                    None
                };
                let trimmed_mean = if stats.trimmed_count > 0 {
                    Some(stats.trimmed_sum / stats.trimmed_count as f64)
                } else {
                    None
                };

                // Winsorized mean
                if let Some(winsorized_mean_val) = winsorized_mean
                    && let Some(idx) = new_column_indices.get(winsorized_col_name.as_str())
                {
                    new_values[*idx] = if field_type.is_date_or_datetime() {
                        days_to_rfc3339(winsorized_mean_val, field_type)
                    } else {
                        util::round_num(winsorized_mean_val, args.flag_round)
                    };
                }

                // Winsorized variance and stddev
                if let Some(winsorized_mean_val) = winsorized_mean
                    && stats.winsorized_count > 1
                {
                    let n = stats.winsorized_count as f64;
                    let winsorized_variance = (stats.sum_squares_winsorized
                        - (stats.winsorized_sum * stats.winsorized_sum / n))
                        / (n - 1.0);
                    if winsorized_variance >= 0.0 {
                        let winsorized_stddev = winsorized_variance.sqrt();
                        if let Some(idx) = new_column_indices.get(&winsorized_stddev_name) {
                            new_values[*idx] = util::round_num(winsorized_stddev, args.flag_round);
                        }
                        if let Some(idx) = new_column_indices.get(&winsorized_variance_name) {
                            new_values[*idx] =
                                util::round_num(winsorized_variance, args.flag_round);
                        }
                        // Winsorized coefficient of variation
                        if winsorized_mean_val.abs() > f64::EPSILON
                            && let Some(idx) = new_column_indices.get(&winsorized_cv_name)
                        {
                            let cv = winsorized_stddev / winsorized_mean_val.abs();
                            new_values[*idx] = util::round_num(cv, args.flag_round);
                        }
                        // Winsorized stddev ratio
                        if let Some(stddev_val) = stddev
                            && stddev_val.abs() > f64::EPSILON
                            && let Some(idx) = new_column_indices.get(&winsorized_stddev_ratio_name)
                        {
                            let ratio = winsorized_stddev / stddev_val;
                            new_values[*idx] = util::round_num(ratio, args.flag_round);
                        }
                    }
                }

                // Winsorized range
                if let (Some(min_winsorized), Some(max_winsorized)) =
                    (stats.min_winsorized, stats.max_winsorized)
                    && let Some(idx) = new_column_indices.get(&winsorized_range_name)
                {
                    let range = max_winsorized - min_winsorized;
                    new_values[*idx] = util::round_num(range, args.flag_round);
                }

                // Trimmed mean
                if let Some(trimmed_mean_val) = trimmed_mean
                    && let Some(idx) = new_column_indices.get(trimmed_col_name.as_str())
                {
                    new_values[*idx] = if field_type.is_date_or_datetime() {
                        days_to_rfc3339(trimmed_mean_val, field_type)
                    } else {
                        util::round_num(trimmed_mean_val, args.flag_round)
                    };
                }

                // Trimmed variance and stddev
                if let Some(trimmed_mean_val) = trimmed_mean
                    && stats.trimmed_count > 1
                {
                    let n = stats.trimmed_count as f64;
                    let trimmed_variance = (stats.sum_squares_trimmed
                        - (stats.trimmed_sum * stats.trimmed_sum / n))
                        / (n - 1.0);
                    if trimmed_variance >= 0.0 {
                        let trimmed_stddev = trimmed_variance.sqrt();
                        if let Some(idx) = new_column_indices.get(&trimmed_stddev_name) {
                            new_values[*idx] = util::round_num(trimmed_stddev, args.flag_round);
                        }
                        if let Some(idx) = new_column_indices.get(&trimmed_variance_name) {
                            new_values[*idx] = util::round_num(trimmed_variance, args.flag_round);
                        }
                        // Trimmed coefficient of variation
                        if trimmed_mean_val.abs() > f64::EPSILON
                            && let Some(idx) = new_column_indices.get(&trimmed_cv_name)
                        {
                            let cv = trimmed_stddev / trimmed_mean_val.abs();
                            new_values[*idx] = util::round_num(cv, args.flag_round);
                        }
                        // Trimmed stddev ratio
                        if let Some(stddev_val) = stddev
                            && stddev_val.abs() > f64::EPSILON
                            && let Some(idx) = new_column_indices.get(&trimmed_stddev_ratio_name)
                        {
                            let ratio = trimmed_stddev / stddev_val;
                            new_values[*idx] = util::round_num(ratio, args.flag_round);
                        }
                    }
                }

                // Trimmed range
                if let (Some(min_trimmed), Some(max_trimmed)) =
                    (stats.min_trimmed, stats.max_trimmed)
                    && let Some(idx) = new_column_indices.get(&trimmed_range_name)
                {
                    let range = max_trimmed - min_trimmed;
                    new_values[*idx] = util::round_num(range, args.flag_round);
                }
            }

            // Write Kurtosis, Gini & Atkinson Index from pre-computed results
            if (new_column_indices.contains_key("kurtosis")
                || new_column_indices.contains_key("gini_coefficient")
                || new_column_indices.contains_key(&atkinson_index_col_name))
                && !field_name.is_empty()
                && let Some(stats) = kga_stats.get(field_name)
            {
                // Kurtosis
                if let Some(kurtosis_val) = stats.kurtosis
                    && let Some(idx) = new_column_indices.get("kurtosis")
                {
                    new_values[*idx] = util::round_num(kurtosis_val, args.flag_round);
                }

                // Gini coefficient
                if let Some(gini_val) = stats.gini_coefficient
                    && let Some(idx) = new_column_indices.get("gini_coefficient")
                {
                    new_values[*idx] = util::round_num(gini_val, args.flag_round);
                }

                // Atkinson Index
                if let Some(atkinson_val) = stats.atkinson_index
                    && let Some(idx) = new_column_indices.get(&atkinson_index_col_name)
                {
                    new_values[*idx] = util::round_num(atkinson_val, args.flag_round);
                }

                // Theil Index
                if let Some(theil_val) = stats.theil_index
                    && let Some(idx) = new_column_indices.get("theil_index")
                {
                    new_values[*idx] = util::round_num(theil_val, args.flag_round);
                }

                // Mean Absolute Deviation from mean
                if let Some(mean_ad_val) = stats.mean_ad
                    && let Some(idx) = new_column_indices.get("mean_ad")
                {
                    new_values[*idx] = util::round_num(mean_ad_val, args.flag_round);
                }
            }
        }
        // Write existing fields + new values directly (avoids record.clone())
        for field in record {
            wtr.write_field(field)?;
        }
        for val in &new_values {
            wtr.write_field(val)?;
        }
        wtr.write_record(None::<&[u8]>)?;
    }

    wtr.flush()?;

    winfo!(
        "Added {} additional statistics columns to {}",
        new_columns.len(),
        output_path.display()
    );
    winfo!("Elapsed: {:.2}s", start_time.elapsed().as_secs_f64());

    // Regenerate the .stats.csv.data.jsonl so downstream "smart" commands
    // (pivotp, schema, etc.) can see moarstats columns
    if let Some(output_path_str) = output_path.to_str() {
        let jsonl_path = PathBuf::from(format!("{output_path_str}.data.jsonl"));
        if let Err(e) = util::csv_to_jsonl(
            output_path_str,
            &crate::cmd::stats::STATSDATA_TYPES_MAP,
            &jsonl_path,
            b',',
        ) {
            wwarn!("Failed to regenerate stats JSONL cache: {e}");
        }
    } else {
        wwarn!(
            "Output path {:?} is not valid UTF-8, skipping JSONL cache regeneration",
            output_path
        );
    }

    // Clean up temporary joined file if it was created
    if let Some(ref temp_path) = temp_joined_path
        && temp_path.exists()
        && let Err(e) = fs::remove_file(temp_path)
    {
        wwarn!(
            "Failed to remove temporary joined file {}: {}",
            temp_path.display(),
            e
        );
    }

    Ok(())
}

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

    #[test]
    fn parse_float_opt_filters_nan_and_infinity() {
        assert_eq!(parse_float_opt(""), None);
        assert_eq!(parse_float_opt("1.5"), Some(1.5));
        assert_eq!(parse_float_opt("NaN"), None);
        assert_eq!(parse_float_opt("nan"), None);
        assert_eq!(parse_float_opt("Infinity"), None);
        assert_eq!(parse_float_opt("-Infinity"), None);
        assert_eq!(parse_float_opt("inf"), None);
        assert_eq!(parse_float_opt("not a number"), None);
    }

    #[test]
    fn parse_float_opt_from_bytes_filters_nan_and_infinity() {
        assert_eq!(parse_float_opt_from_bytes(b""), None);
        assert_eq!(parse_float_opt_from_bytes(b"42"), Some(42.0));
        assert_eq!(parse_float_opt_from_bytes(b"NaN"), None);
        assert_eq!(parse_float_opt_from_bytes(b"inf"), None);
    }

    #[test]
    fn compute_pearson_skewness_handles_zero_stddev() {
        // Zero stddev -> None (avoid divide-by-zero)
        assert_eq!(
            compute_pearson_skewness(Some(5.0), Some(5.0), Some(0.0)),
            None,
        );
        // Any None input -> None
        assert_eq!(compute_pearson_skewness(None, Some(5.0), Some(1.0)), None);
        // Normal case: 3 * (mean - median) / stddev
        assert_eq!(
            compute_pearson_skewness(Some(10.0), Some(8.0), Some(2.0)),
            Some(3.0),
        );
    }

    #[test]
    fn compute_kendall_tau_no_nan_on_all_ties() {
        // All identical values: every pair is a tie in both x and y.
        // Pre-fix this would produce sqrt(NaN) from rounding making
        // (n0 - n1) or (n0 - n2) slightly negative.
        let x = vec![1.0; 10];
        let y = vec![2.0; 10];
        let tau = compute_kendall_tau(&x, &y);
        // Denominator is zero -> None, never NaN.
        assert!(tau.is_none());
    }

    #[test]
    fn compute_kendall_tau_perfect_concordance() {
        let x: Vec<f64> = (1..=5).map(f64::from).collect();
        let y: Vec<f64> = (1..=5).map(f64::from).collect();
        assert_eq!(compute_kendall_tau(&x, &y), Some(1.0));
    }

    #[test]
    fn compute_normalized_mutual_information_epsilon_guard() {
        // h_x = 0 -> None (early guard)
        assert_eq!(
            compute_normalized_mutual_information(Some(0.5), Some(0.0), Some(1.0)),
            None,
        );
        // Very small positive entropies that would produce a subnormal
        // denominator -> None (epsilon guard, not float-equality).
        let tiny = 1e-200;
        assert_eq!(
            compute_normalized_mutual_information(Some(tiny), Some(tiny), Some(tiny)),
            None,
        );
    }

    #[test]
    fn outlier_count_indices_are_in_range() {
        // Guard against a refactor changing these without updating COUNTS_LEN.
        assert!(OUTLIER_EXTREME_LOWER < OUTLIER_COUNTS_LEN);
        assert!(OUTLIER_MILD_LOWER < OUTLIER_COUNTS_LEN);
        assert!(OUTLIER_NORMAL < OUTLIER_COUNTS_LEN);
        assert!(OUTLIER_MILD_UPPER < OUTLIER_COUNTS_LEN);
        assert!(OUTLIER_EXTREME_UPPER < OUTLIER_COUNTS_LEN);
        assert!(OUTLIER_TOTAL < OUTLIER_COUNTS_LEN);
    }

    // ---------------------------------------------------------------------
    // Pure helper-function tests. These guard the math kernels exercised
    // indirectly by the integration suite — a regression in one of these
    // would otherwise surface only as a numeric drift in a large CSV diff.
    // ---------------------------------------------------------------------

    #[test]
    fn fmt_pct_drops_fraction_for_integral_values() {
        assert_eq!(fmt_pct(5.0), "5");
        assert_eq!(fmt_pct(25.0), "25");
        assert_eq!(fmt_pct(0.0), "0");
        // Non-integral values retain their fractional part via Display.
        assert_eq!(fmt_pct(5.5), "5.5");
        assert_eq!(fmt_pct(2.5), "2.5");
    }

    #[test]
    fn compute_quartile_coefficient_dispersion_basic_and_edges() {
        // Standard case: (3 - 1) / (3 + 1) = 0.5
        assert_eq!(
            compute_quartile_coefficient_dispersion(Some(1.0), Some(3.0)),
            Some(0.5),
        );
        // Q1 == Q3 -> invalid order -> None (would otherwise be 0/(2*Q)).
        assert_eq!(
            compute_quartile_coefficient_dispersion(Some(2.0), Some(2.0)),
            None,
        );
        // Q1 > Q3 -> invalid order -> None.
        assert_eq!(
            compute_quartile_coefficient_dispersion(Some(5.0), Some(3.0)),
            None,
        );
        // Q1 == -Q3 (sum near zero) -> None to avoid divide-by-near-zero.
        assert_eq!(
            compute_quartile_coefficient_dispersion(Some(-2.0), Some(2.0)),
            None,
        );
        // Either operand None -> None.
        assert_eq!(
            compute_quartile_coefficient_dispersion(None, Some(3.0)),
            None,
        );
        assert_eq!(
            compute_quartile_coefficient_dispersion(Some(1.0), None),
            None,
        );
    }

    #[test]
    fn compute_robust_cv_basic_and_zero_median() {
        assert_eq!(compute_robust_cv(Some(2.0), Some(4.0)), Some(0.5));
        // |median| < EPSILON -> None.
        assert_eq!(compute_robust_cv(Some(2.0), Some(0.0)), None);
        // Negative median uses |median|, so still positive.
        assert_eq!(compute_robust_cv(Some(2.0), Some(-4.0)), Some(0.5));
        assert_eq!(compute_robust_cv(None, Some(4.0)), None);
        assert_eq!(compute_robust_cv(Some(2.0), None), None);
    }

    #[test]
    fn compute_normalized_entropy_handles_low_cardinality() {
        // cardinality 0 or 1 -> always 0.0 (degenerate distribution).
        assert_eq!(compute_normalized_entropy(Some(0.0), Some(0)), Some(0.0));
        assert_eq!(compute_normalized_entropy(Some(0.0), Some(1)), Some(0.0));
        // Uniform distribution over k values: entropy == log2(k), so
        // normalized entropy == 1.
        let k: u64 = 8;
        let entropy = (k as f64).log2();
        let got = compute_normalized_entropy(Some(entropy), Some(k)).unwrap();
        assert!((got - 1.0).abs() < 1e-12, "expected ~1.0, got {got}");
        // Either input None -> None.
        assert_eq!(compute_normalized_entropy(None, Some(8)), None);
        assert_eq!(compute_normalized_entropy(Some(1.0), None), None);
    }

    #[test]
    fn compute_bimodality_coefficient_basic_and_singularity() {
        // (skew^2 + 1) / (kurt + 3) for skew=0, kurt=0 -> 1/3.
        let got = compute_bimodality_coefficient(Some(0.0), Some(0.0)).unwrap();
        assert!((got - 1.0 / 3.0).abs() < 1e-12);
        // kurt == -3 makes the denominator zero -> None.
        assert_eq!(compute_bimodality_coefficient(Some(0.0), Some(-3.0)), None,);
        assert_eq!(compute_bimodality_coefficient(None, Some(0.0)), None);
        assert_eq!(compute_bimodality_coefficient(Some(0.0), None), None);
    }

    #[test]
    fn compute_jarque_bera_small_n_and_known_values() {
        // The `kurtosis` parameter here is *excess* kurtosis (i.e. raw - 3),
        // matching the convention produced by qsv's stats command. The hand-
        // computed expectations below silently encode that — if the function
        // ever switches to raw kurtosis, the second case will break.
        // n < 3 -> None.
        assert_eq!(compute_jarque_bera(Some(0.0), Some(0.0), 0), None);
        assert_eq!(compute_jarque_bera(Some(0.0), Some(0.0), 2), None);
        // Skew = 0, excess kurt = 0 -> JB = 0, p = exp(0) = 1 (normal-looking moments).
        let (jb, p) = compute_jarque_bera(Some(0.0), Some(0.0), 100).unwrap();
        assert!(jb.abs() < 1e-12);
        assert!((p - 1.0).abs() < 1e-12);
        // Hand-computed (excess kurtosis convention):
        //   n = 60, skew = 1, excess kurt = 2 -> JB = (60/6) * (1 + 4/4) = 20, p = exp(-10).
        let (jb, p) = compute_jarque_bera(Some(1.0), Some(2.0), 60).unwrap();
        assert!((jb - 20.0).abs() < 1e-9);
        assert!((p - (-10.0_f64).exp()).abs() < 1e-12);
        // Either moment None -> None.
        assert_eq!(compute_jarque_bera(None, Some(0.0), 100), None);
        assert_eq!(compute_jarque_bera(Some(0.0), None, 100), None);
    }

    #[test]
    fn merge_correlation_states_zero_count_passthrough() {
        let empty = CorrelationState::default();
        let mut populated = CorrelationState::default();
        update_correlation_state(&mut populated, 1.0, 2.0);
        update_correlation_state(&mut populated, 3.0, 4.0);

        // Merging with an empty state must return the populated state untouched.
        let merged_a = merge_correlation_states(&empty, &populated);
        assert_eq!(merged_a.count, populated.count);
        assert!((merged_a.mean_x - populated.mean_x).abs() < 1e-12);
        assert!((merged_a.cxy - populated.cxy).abs() < 1e-12);

        let merged_b = merge_correlation_states(&populated, &empty);
        assert_eq!(merged_b.count, populated.count);
        assert!((merged_b.mean_x - populated.mean_x).abs() < 1e-12);
        assert!((merged_b.cxy - populated.cxy).abs() < 1e-12);
    }

    #[test]
    fn merge_correlation_states_matches_single_pass() {
        // Build the "ground truth" state by feeding all pairs sequentially.
        let xs = [1.0_f64, 2.0, 4.0, 7.0, 11.0, 16.0, 22.0, 29.0];
        let ys = [3.0_f64, 1.0, 4.0, 1.0, 5.0, 9.0, 2.0, 6.0];

        let mut full = CorrelationState::default();
        for i in 0..xs.len() {
            update_correlation_state(&mut full, xs[i], ys[i]);
        }

        // Build two partial states for a non-trivial split (3 + 5).
        let split = 3;
        let mut part1 = CorrelationState::default();
        for i in 0..split {
            update_correlation_state(&mut part1, xs[i], ys[i]);
        }
        let mut part2 = CorrelationState::default();
        for i in split..xs.len() {
            update_correlation_state(&mut part2, xs[i], ys[i]);
        }

        let merged = merge_correlation_states(&part1, &part2);

        // Counts must match exactly; floats within tight tolerance because
        // the Welford parallel formula is algebraically equivalent but
        // takes a different rounding path than single-pass.
        assert_eq!(merged.count, full.count);
        assert!((merged.mean_x - full.mean_x).abs() < 1e-9);
        assert!((merged.mean_y - full.mean_y).abs() < 1e-9);
        assert!((merged.m2_x - full.m2_x).abs() < 1e-9);
        assert!((merged.m2_y - full.m2_y).abs() < 1e-9);
        assert!((merged.cxy - full.cxy).abs() < 1e-9);
    }

    #[test]
    fn count_inversions_merge_known_cases() {
        // Helper that runs the function on a fresh buffer pair. Empty input
        // returns 0 directly so the helper itself can be exercised on `&[]`
        // without underflow when computing `last = data.len() - 1`.
        fn count(pairs: &[(f64, f64)]) -> i64 {
            if pairs.is_empty() {
                return 0;
            }
            let mut data = pairs.to_vec();
            let mut temp = vec![(0.0, 0.0); data.len()];
            let last = data.len() - 1;
            count_inversions_merge(&mut data, &mut temp, 0, last)
        }

        // Empty slice -> helper short-circuits to 0.
        assert_eq!(count(&[]), 0);
        // Already sorted by y -> 0 inversions.
        assert_eq!(count(&[(1.0, 1.0), (2.0, 2.0), (3.0, 3.0)]), 0);
        // Reverse-sorted by y -> n*(n-1)/2 inversions for n=3 -> 3.
        assert_eq!(count(&[(1.0, 3.0), (2.0, 2.0), (3.0, 1.0)]), 3);
        // Ties don't count as inversions (uses Ordering::Greater).
        assert_eq!(count(&[(1.0, 5.0), (2.0, 5.0), (3.0, 5.0)]), 0);
        // Mixed: only (3 > 1) and (3 > 2) are inversions for y = [3, 1, 2] -> 2.
        assert_eq!(count(&[(1.0, 3.0), (2.0, 1.0), (3.0, 2.0)]), 2);
    }

    #[test]
    fn parse_date_and_days_to_rfc3339_roundtrip() {
        // Date round-trip: parse "2022-01-15", format with TDate -> "2022-01-15".
        let days = parse_date_to_days("2022-01-15", false).unwrap();
        assert_eq!(days_to_rfc3339(days, FieldType::TDate), "2022-01-15");

        // DateTime round-trip with explicit UTC: converting through `f64` days can
        // introduce tiny rounding differences, so assert that the reconstructed
        // timestamp is within 1 ms rather than requiring exact RFC3339 string
        // equality.
        let dt = "2022-01-15T12:30:45+00:00";
        let dt_days = parse_date_to_days(dt, false).unwrap();
        let dt_rfc3339 = days_to_rfc3339(dt_days, FieldType::TDateTime);
        let reparsed_days = parse_date_to_days(&dt_rfc3339, false).unwrap();
        let one_millisecond_in_days = 1.0 / 86_400_000.0;
        assert!(
            (reparsed_days - dt_days).abs() <= one_millisecond_in_days,
            "expected `{dt_rfc3339}` to round-trip within 1 ms of `{dt}`"
        );

        // Empty input -> None.
        assert_eq!(parse_date_to_days("", false), None);
        // Garbage input -> None (no panic).
        assert_eq!(parse_date_to_days("not-a-date", false), None);
    }

    #[test]
    fn finalize_covariance_sample_and_population() {
        // Build a state for x = y = [1, 2, 3, 4, 5].
        // Centered: each (xi - mean) * (yi - mean) summed = m2 = 10.0
        let mut state = CorrelationState::default();
        for v in [1.0, 2.0, 3.0, 4.0, 5.0] {
            update_correlation_state(&mut state, v, v);
        }
        // Sample covariance: cxy / (n - 1) = 10 / 4 = 2.5.
        let sample = finalize_covariance(&state, true).unwrap();
        assert!((sample - 2.5).abs() < 1e-12, "sample covariance: {sample}");
        // Population covariance: cxy / n = 10 / 5 = 2.0.
        let pop = finalize_covariance(&state, false).unwrap();
        assert!((pop - 2.0).abs() < 1e-12, "population covariance: {pop}");

        // count < 2 -> None.
        let mut tiny = CorrelationState::default();
        update_correlation_state(&mut tiny, 1.0, 2.0);
        assert_eq!(finalize_covariance(&tiny, true), None);
        assert_eq!(finalize_covariance(&tiny, false), None);
    }

    #[test]
    fn finalize_pearson_correlation_guards() {
        // count < 2 -> None.
        let empty = CorrelationState::default();
        assert_eq!(finalize_pearson_correlation(&empty), None);

        // Constant y (variance_y == 0) -> None.
        let mut const_y = CorrelationState::default();
        for x in 1..=5 {
            update_correlation_state(&mut const_y, f64::from(x), 7.0);
        }
        assert_eq!(finalize_pearson_correlation(&const_y), None);
    }

    #[test]
    fn compute_pearson_correlation_known_values() {
        // Perfect positive linear -> +1.
        let r =
            compute_pearson_correlation(&[1.0, 2.0, 3.0, 4.0, 5.0], &[2.0, 4.0, 6.0, 8.0, 10.0])
                .unwrap();
        assert!((r - 1.0).abs() < 1e-12, "perfect positive: {r}");

        // Perfect negative linear -> -1.
        let r =
            compute_pearson_correlation(&[1.0, 2.0, 3.0, 4.0, 5.0], &[10.0, 8.0, 6.0, 4.0, 2.0])
                .unwrap();
        assert!((r + 1.0).abs() < 1e-12, "perfect negative: {r}");

        // Hand-computed: x=[1,2,3,4,5], y=[2,4,5,4,5]
        //   dx = [-2,-1,0,1,2], dy = [-2,0,1,0,1]
        //   Sxy = 6, Sxx = 10, Syy = 6 -> r = 6 / sqrt(60) = sqrt(0.6).
        let r = compute_pearson_correlation(&[1.0, 2.0, 3.0, 4.0, 5.0], &[2.0, 4.0, 5.0, 4.0, 5.0])
            .unwrap();
        assert!(
            (r - 0.6_f64.sqrt()).abs() < 1e-12,
            "fractional case: got {r}, expected {}",
            0.6_f64.sqrt()
        );

        // Mismatched lengths -> None.
        assert_eq!(
            compute_pearson_correlation(&[1.0, 2.0], &[1.0, 2.0, 3.0]),
            None,
        );
        // Length < 2 -> None.
        assert_eq!(compute_pearson_correlation(&[1.0], &[2.0]), None);
        // Constant input (zero variance) -> None, not NaN.
        assert_eq!(
            compute_pearson_correlation(&[5.0, 5.0, 5.0], &[1.0, 2.0, 3.0]),
            None,
        );
    }

    #[test]
    fn compute_spearman_correlation_handles_monotonic_and_ties() {
        // Strictly increasing non-linear (cubic) -> Spearman = +1
        // even though Pearson would be < 1. This is the whole point of
        // rank correlation.
        let r = compute_spearman_correlation(
            &[1.0, 2.0, 3.0, 4.0, 5.0],
            &[1.0, 8.0, 27.0, 64.0, 125.0],
        )
        .unwrap();
        assert!((r - 1.0).abs() < 1e-12, "cubic monotonic: {r}");

        // Strictly decreasing -> -1.
        let r =
            compute_spearman_correlation(&[1.0, 2.0, 3.0, 4.0, 5.0], &[5.0, 4.0, 3.0, 2.0, 1.0])
                .unwrap();
        assert!((r + 1.0).abs() < 1e-12, "decreasing: {r}");

        // Tie handling. y has two pairs of ties at 4 and 5:
        //   x=[1,2,3,4,5] -> ranks [1,2,3,4,5]
        //   y=[2,4,5,4,5] -> 2->rank 1, 4 (×2)->avg rank 2.5, 5 (×2)->avg rank 4.5
        //   y_ranks = [1, 2.5, 4.5, 2.5, 4.5]
        // Pearson on the ranks: dx = [-2,-1,0,1,2], dy = [-2,-0.5,1.5,-0.5,1.5]
        //   Sxy = 7, Sxx = 10, Syy = 9 -> r = 7 / sqrt(90).
        let r =
            compute_spearman_correlation(&[1.0, 2.0, 3.0, 4.0, 5.0], &[2.0, 4.0, 5.0, 4.0, 5.0])
                .unwrap();
        let expected = 7.0_f64 / 90.0_f64.sqrt();
        assert!(
            (r - expected).abs() < 1e-9,
            "tied Spearman: got {r}, expected {expected}",
        );

        // Length guards mirror Pearson.
        assert_eq!(
            compute_spearman_correlation(&[1.0, 2.0], &[1.0, 2.0, 3.0]),
            None,
        );
        assert_eq!(compute_spearman_correlation(&[1.0], &[2.0]), None);
    }
}