#![allow(clippy::missing_asserts_for_indexing)]
static USAGE: &str = r#"
Compute a frequency distribution table on input data. It has CSV and JSON output modes.
https://en.wikipedia.org/wiki/Frequency_(statistics)#Frequency_distribution_table

In CSV output mode (default), the table is formatted as CSV data with the following
columns - field,value,count,percentage,rank.

The rank column is 1-based and is calculated based on the count of the values,
with the most frequent having a rank of 1. In case of ties, the rank is calculated
based on the rank-strategy option - "dense" (default), "min", "max", "ordinal", or "average".

Only the top N values (set by the --limit option) are computed, with the rest of the values
grouped into an "Other" category with a special rank of 0. The "Other" category includes
the count of remaining unique values that are not in the top N values.

In JSON output mode, the table is formatted as nested JSON data. In addition to
the columns above, the JSON output also includes the row count, field count, rank-strategy,
each field's data type, cardinality, nullcount, sparsity, uniqueness_ratio and its stats.

Since this command computes an exact frequency distribution table, memory proportional
to the cardinality of each column would be normally required.

However, this is problematic for columns with ALL unique values (e.g. an ID column),
as the command will need to allocate memory proportional to the column's cardinality.

To overcome this, the frequency command uses several mechanisms:

STATS CACHE:
If the stats cache exists for the input file, it is used to get column cardinality information.
This short-circuits frequency compilation for columns with all unique values (i.e. where
rowcount == cardinality), eliminating the need to maintain an in-memory hashmap for ID columns.
This allows `frequency` to handle larger-than-memory datasets with the added benefit of also
making it faster when working with datasets with ID columns.

That's why for MAXIMUM PERFORMANCE, it's HIGHLY RECOMMENDED to create an index (`qsv index data.csv`)
and pre-populate the stats cache (`qsv stats data.csv --cardinality --stats-jsonl`)
BEFORE running `frequency`.

MEMORY-AWARE CHUNKING:
When working with large datasets, memory-aware chunking is automatically enabled. Chunk size
is dynamically calculated based on available memory and record sampling.

You can override this behavior by setting the QSV_FREQ_CHUNK_MEMORY_MB environment variable.
(set to 0 for dynamic sizing, or a positive number for a fixed memory limit per chunk,
or any non-u64 value (e.g. -1 or "auto") for CPU-based chunking (1 chunk = num records/number of
CPUs)), or by setting the --jobs option.

"COMPLETE" FREQUENCY TABLES FOR ID COLUMNS:
By default, ID columns will have an `<ALL UNIQUE>` value with count equal to
rowcount and percentage set to 100 with a rank of 0. This is done by using the
stats cache to fetch each column's cardinality - allowing qsv to short-circuit
frequency compilation and eliminate the need to maintain a hashmap for ID columns.

If you wish to compile a "complete" frequency table even for ID columns, set
QSV_STATSCACHE_MODE to "none". This will force the frequency command to compute
frequencies for all columns regardless of cardinality, even for ID columns.

In this case, the unique limit (--unq-limit) option is particularly useful when
a column has all unique values  and --limit is set to 0.
Without a unique limit, the frequency table for that column will be the same as
the number of rows in the data.
With a unique limit, the frequency table will be a sample of N unique values,
all with a count of 1.

The --lmt-threshold option also allows you to apply the --limit and --unq-limit
options only when the number of unique items in a column >= threshold.
This is useful when you want to apply limits only to columns with a large number
of unique items and not to columns with a small number of unique items.

For examples, see https://github.com/dathere/qsv/blob/master/tests/test_frequency.rs.
See also https://github.com/dathere/qsv/wiki/Aggregation-and-Statistics#frequency

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

frequency options:
    -s, --select <arg>      Select a subset of columns to compute frequencies
                            for. See 'qsv select --help' for the format
                            details. This is provided here because piping 'qsv
                            select' into 'qsv frequency' will disable the use
                            of indexing.
    -l, --limit <arg>       Limit the frequency table to the N most common
                            items. Set to '0' to disable a limit.
                            If negative, only return values with an occurrence
                            count >= absolute value of the negative limit.
                            e.g. --limit -2 will only return values with an
                            occurrence count >= 2.
                            [default: 10]
    --sketch-method <m>     Algorithm used to compute the frequency table.
                            Choices: 'exact' (default) tracks every distinct value
                            in a HashMap; 'frequent_items' uses the Apache
                            DataSketches Frequent Items (Misra-Gries) sketch to
                            track top-K heavy hitters with bounded error and
                            constant memory. The frequent_items mode rejects
                            asc, weight, ignore-case, no-trim, other-sorted,
                            null-sorted, frequency-jsonl, stats-filter, and
                            json/pretty-json/toon output; the frequency cache is
                            bypassed. Counts are estimates. The flags
                            rank-strategy, lmt-threshold, and unq-limit are
                            silently ignored under this mode (the sketch's
                            natural ordering is top-K by estimate descending;
                            tied counts use the sketch's hash-table iteration
                            order). The 'Other' row format diverges from exact:
                            label is the bare other-text (no unique-count suffix
                            since the sketch cannot recover the true count of
                            items not in the top-K); rank is 0 to match the
                            exact convention.
                            Note: 'frequent_items' requires a little-endian
                            target. Apache DataSketches does not support
                            big-endian platforms (e.g., s390x); on those
                            builds, this choice is rejected.
                            [default: exact]
    --sketch-map-size <n>   Maximum map size for the Frequent Items sketch.
                            Must be a power of two and at least 8. Larger values
                            tighten error bounds at the cost of more memory.
                            Only used when sketch-method is frequent_items.
                            [default: 4096]
    -u, --unq-limit <arg>   If a column has all unique values, limit the
                            frequency table to a sample of N unique items.
                            Set to '0' to disable a unique_limit.
                            Only applies in unweighted mode; ignored when --weight is set.
                            [default: 10]
    --lmt-threshold <arg>   The threshold for which --limit and --unq-limit
                            will be applied. If the number of unique items
                            in a column >= threshold, the limits will be applied.
                            Set to '0' to disable the threshold and always apply limits.
                            [default: 0]
-r, --rank-strategy <arg>   The strategy to use when there are count-tied values in the frequency table.
                            See https://en.wikipedia.org/wiki/Ranking for more info.
                            Valid values are:
                              * dense: Assigns consecutive integers regardless of ties,
                                incrementing by 1 for each new count value (AKA "1223" ranking).
                              * min: Tied items receive the minimum rank position (AKA "1224" ranking).
                              * max: Tied items receive the maximum rank position (AKA "1334" ranking).
                              * ordinal: The next rank is the current rank plus 1 (AKA "1234" ranking).
                              * average: Tied items receive the average of their ordinal positions
                                (AKA "1 2.5 2.5 4" ranking).
                            Note that tied values with the same rank are sorted alphabetically.
                            [default: dense]
    --pct-dec-places <arg>  The number of decimal places to round the percentage to.
                            If negative, the number of decimal places will be set
                            automatically to the minimum number of decimal places needed
                            to represent the percentage accurately, up to the absolute
                            value of the negative number.
                            [default: -5]
    --other-sorted          By default, the "Other" category is placed at the
                            end of the frequency table for a field. If this is enabled, the
                            "Other" category will be sorted with the rest of the
                            values by count.
    --other-text <arg>      The text to use for the "Other" category. If set to the
                            literal string "<NONE>" (case-sensitive, exact match), the
                            "Other" category will not be included in the frequency table.
                            [default: Other]
    --no-other              Don't include the "Other" category in the frequency table.
                            This is equivalent to --other-text "<NONE>".
    --null-sorted           By default, the NULL category (controlled by --null-text)
                            is placed at the end of the frequency table for a field,
                            after "Other" if present. If this is enabled, the NULL
                            category will be sorted with the rest of the values by count.
    -a, --asc               Sort the frequency tables in ascending order by count.
                            The default is descending order. Note that this option will
                            also reverse ranking - i.e. the LEAST frequent values will
                            have a rank of 1.
    --no-trim               Don't trim whitespace from values when computing frequencies.
                            The default is to trim leading and trailing whitespaces.
    --null-text <arg>       The text to use for NULL values. If set to the literal
                            string "<NONE>" (case-sensitive, exact match), NULLs
                            will not be included in the frequency table
                            (equivalent to --no-nulls).
                            [default: (NULL)]
    --no-nulls              Don't include NULLs in the frequency table.
                            This is equivalent to --null-text "<NONE>".
    --pct-nulls             Include NULL values in percentage and rank calculations.
                            When disabled (default), percentages are "valid percentages"
                            calculated with NULLs excluded from the denominator, and
                            NULL entries display empty percentage and rank values.
                            When enabled, NULLs are included in the denominator
                            (original behavior).
                            Has no effect when --no-nulls is set.
    -i, --ignore-case       Ignore case when computing frequencies.
    --no-float <cols>       Exclude Float columns from frequency analysis.
                            Floats typically contain continuous values where
                            frequency tables are not meaningful.
                            To exclude ALL Float columns, use --no-float "*"
                            To exclude Floats except specific columns, specify
                            a comma-separated list of Float columns to INCLUDE.
                            e.g. "--no-float *" excludes all Floats
                                 "--no-float price,rate" excludes Floats
                                    except 'price' and 'rate'
                            Requires stats cache for type detection.
    --stats-filter <expr>   Filter columns based on their statistics using a Luau expression.
                            Columns where the expression evaluates to `true` are EXCLUDED.
                            Available fields: field, type, is_ascii, cardinality, nullcount,
                            sum, min, max, range, sort_order, min_length, max_length, mean,
                            stddev, variance, cv, sparsity, q1, q2_median, q3, iqr, mad,
                            skewness, mode, antimode, n_negative, n_zero, n_positive, etc.
                            e.g. "nullcount > 1000" - exclude columns with many nulls
                                 "type == 'Float'" - exclude Float columns
                                 "cardinality > 500 and nullcount > 0" - compound expression
                            Requires stats cache and the "luau" feature.
   --all-unique-text <arg>  The text to use for the "<ALL_UNIQUE>" category.
                            [default: <ALL_UNIQUE>]
    --vis-whitespace        Visualize whitespace characters in the output. See
                            https://github.com/dathere/qsv/wiki/Aggregation-and-Statistics#whitespace-markers
                            for the list of whitespace markers.
    -j, --jobs <arg>        The number of jobs to run in parallel when the given CSV data has
                            an index. Note that a file handle is opened for each job.
                            When not set, defaults to the number of CPUs detected.

                            FREQUENCY CACHE OPTIONS:
    --frequency-jsonl       Write the complete frequency distribution as a
                            JSONL cache file (FILESTEM.freq.csv.data.jsonl).
                            Requires a non-stdin input file. The cache contains
                            metadata and per-column frequency data.
                            ALL_UNIQUE columns (rowcount == cardinality) get a single
                            ALL_UNIQUE sentinel. HIGH_CARDINALITY columns (cardinality
                            exceeds the smaller of --high-card-threshold/--high-card-pct
                            of rowcount) get a single HIGH_CARDINALITY sentinel.
                            When a valid (fresh) cache already exists, frequency will
                            automatically reuse it instead of recomputing from the CSV.
                            Use --force to regenerate the cache even when it is valid.
                            Cache is NOT used when --ignore-case, --no-trim, or --weight
                            are active, as these change how values are computed.
    --high-card-threshold <arg>  Absolute cardinality threshold for HIGH_CARDINALITY
                                 classification in the frequency cache.
                                 Can also be set with QSV_FREQ_HIGH_CARD_THRESHOLD env var
                                 (env var takes precedence when CLI value equals the default).
                                 Only used with --frequency-jsonl.
                                 [default: 100]
    --high-card-pct <arg>   Percentage of rowcount threshold for HIGH_CARDINALITY
                            classification in the frequency cache. Must be between 1 and 100.
                            Can also be set with QSV_FREQ_HIGH_CARD_PCT env var
                            (env var takes precedence when CLI value equals the default).
                            Only used with --frequency-jsonl.
                            [default: 90]
    --force                 Force recomputation even when a valid frequency cache
                            exists, bypassing the auto-reuse path. Also regenerates
                            the cache when combined with --frequency-jsonl.

                            JSON OUTPUT OPTIONS:
    --json                  Output frequency table as nested JSON instead of CSV.
                            The JSON output includes additional metadata: row count, field count,
                            data type, cardinality, null count, sparsity, uniqueness_ratio and
                            17 additional stats (e.g. sum, min, max, range, sort_order, mean, sem, etc.).
    --pretty-json           Same as --json but pretty prints the JSON output.
    --toon                  Output frequency table and select stats in TOON format instead of CSV.
                            TOON is a compact, human-readable encoding of the JSON data model for LLM prompts.
                            See https://toonformat.dev/ for more info.
    --no-stats              When using the JSON or TOON output mode, do not include the additional stats.
    --weight <column>       Compute weighted frequencies using the specified column as weights.
                            The weight column must be numeric. When specified, frequency counts
                            are multiplied by the weight value for each row. The weight column is
                            automatically excluded from frequency computation. Missing or
                            unparsable weights default to 1.0. Zero, negative, NaN and infinite
                            weights are ignored and do not contribute to frequencies.

Common options:
    -h, --help             Display this message
    -o, --output <file>    Write output to <file> instead of stdout.
    -n, --no-headers       When set, the first row will NOT be included
                           in the frequency table. Additionally, the 'field'
                           column will be 1-based indices instead of header
                           names.
    -d, --delimiter <arg>  The field delimiter for reading CSV data.
                           Must be a single character. (default: ,)
    --memcheck             Use CONSERVATIVE heuristics for the in-memory load
                           check (file size vs. available + free_swap × platform
                           factor − headroom), instead of the default NORMAL
                           check (file size vs. total memory − headroom). The
                           CONSERVATIVE check is stricter and trips OOM far
                           more readily. (See also: QSV_MEMORY_CHECK env var,
                           equivalent to passing --memcheck.)
                           Independently of this flag, the in-memory load
                           check runs whenever frequency takes the
                           non-parallel path. On OOM (in either NORMAL or
                           CONSERVATIVE mode), qsv auto-creates an index when
                           no index exists (skipped for stdin) AND switches
                           to the Frequent Items sketch (Apache DataSketches
                           Misra-Gries, equivalent to `--sketch-method
                           frequent_items`) where compatible. The sketch
                           fallback can also fire when an index is already
                           present and the OOM still trips (e.g., when jobs
                           is pinned to 1 on a pre-indexed file). A wwarn is
                           emitted when the sketch fallback engages.
"#;

use core::hint::cold_path;
use std::{fs, io, str::FromStr, sync::OnceLock};

use crossbeam_channel;
use foldhash::{HashMap, HashMapExt, HashSet, HashSetExt};
use indicatif::HumanCount;
use rayon::prelude::*;
use rust_decimal::prelude::*;
use serde::{Deserialize, Serialize};
use serde_json::{self, Value as JsonValue};
use stats::{Commute, Frequencies};
use threadpool::ThreadPool;
use toon_format::{EncodeOptions, encode};

use crate::{
    CliResult,
    cmd::stats::StatsData,
    config::{Config, Delimiter},
    index::Indexed,
    select::{SelectColumns, Selection},
    util::{self, ByteString, StatsMode, get_stats_records},
};

#[derive(Clone, Copy, Debug, Deserialize, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum RankStrategy {
    Min,
    Max,
    Dense,
    Ordinal,
    Average,
}

impl FromStr for RankStrategy {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s.to_lowercase().as_str() {
            "min" => Ok(RankStrategy::Min),
            "max" => Ok(RankStrategy::Max),
            "dense" => Ok(RankStrategy::Dense),
            "ordinal" => Ok(RankStrategy::Ordinal),
            "average" => Ok(RankStrategy::Average),
            _ => Err(format!(
                "Invalid rank-strategy: '{s}'. Valid values are: dense, min, max, ordinal, average"
            )),
        }
    }
}

#[allow(clippy::unsafe_derive_deserialize)]
#[derive(Clone, Deserialize)]
pub struct Args {
    pub arg_input:                Option<String>,
    pub flag_select:              SelectColumns,
    pub flag_limit:               isize,
    pub flag_unq_limit:           usize,
    pub flag_lmt_threshold:       usize,
    pub flag_rank_strategy:       RankStrategy,
    pub flag_pct_dec_places:      isize,
    pub flag_other_sorted:        bool,
    pub flag_other_text:          String,
    pub flag_no_other:            bool,
    pub flag_null_sorted:         bool,
    pub flag_asc:                 bool,
    pub flag_no_trim:             bool,
    pub flag_null_text:           String,
    pub flag_no_nulls:            bool,
    pub flag_pct_nulls:           bool,
    pub flag_ignore_case:         bool,
    pub flag_no_float:            Option<String>,
    #[cfg(feature = "luau")]
    pub flag_stats_filter:        Option<String>,
    pub flag_all_unique_text:     String,
    pub flag_jobs:                Option<usize>,
    pub flag_output:              Option<String>,
    pub flag_no_headers:          bool,
    pub flag_delimiter:           Option<Delimiter>,
    pub flag_memcheck:            bool,
    pub flag_vis_whitespace:      bool,
    pub flag_frequency_jsonl:     bool,
    pub flag_high_card_threshold: usize,
    pub flag_high_card_pct:       u8,
    pub flag_force:               bool,
    pub flag_json:                bool,
    pub flag_pretty_json:         bool,
    pub flag_toon:                bool,
    pub flag_no_stats:            bool,
    pub flag_weight:              Option<String>,
    pub flag_sketch_method:       String,
    pub flag_sketch_map_size:     usize,
}

const NON_UTF8_ERR: &str = "<Non-UTF8 ERROR>";

// Name-keyed stats records. Retained ONLY for lookups that legitimately address a
// column by its user-specified name: the `--weight` column's tolerance scale and
// `--stats-filter`. NOT used for per-output-column JSON stats (see STATS_RECORDS_BY_POS),
// which must be positional so duplicate-named columns don't collapse onto one record.
static STATS_RECORDS: OnceLock<HashMap<String, StatsData>> = OnceLock::new();
// Per-output-column stats for JSON/TOON output, aligned positionally to the FINAL
// selected columns (index = output column position). Positional — not name-keyed — so
// duplicate-named columns each report their own stats.
static STATS_RECORDS_BY_POS: OnceLock<Vec<StatsData>> = OnceLock::new();
static NULL_VAL: OnceLock<Vec<u8>> = OnceLock::new();
static UNIQUE_COLUMNS_VEC: OnceLock<Vec<usize>> = OnceLock::new();
// Cardinalities aligned positionally to the FINAL selected columns (NOT keyed by
// name). Name-keying collapsed duplicate-named columns onto the first match; using
// position lets each duplicate keep its own cardinality.
static COL_CARDINALITY_VEC: OnceLock<Vec<u64>> = OnceLock::new();
static FREQ_ROW_COUNT: OnceLock<u64> = OnceLock::new();
static FLOAT_COLUMNS_TO_SKIP: OnceLock<Vec<usize>> = OnceLock::new();
#[cfg(feature = "luau")]
static STATS_FILTER_COLUMNS_TO_SKIP: OnceLock<Vec<usize>> = OnceLock::new();
static EMPTY_VEC: Vec<u64> = Vec::new();
static EMPTY_USIZE_VEC: Vec<usize> = Vec::new();
static ALL_UNIQUE_TEXT: OnceLock<Vec<u8>> = OnceLock::new();
// Partial frequency cache: when some columns are cached but others need computation,
// FREQ_CACHE_SKIP marks which columns to skip in ftables_unweighted (true = cached),
// and FREQ_CACHE_FTABLES holds their pre-built FTables for merging after computation.
static FREQ_CACHE_SKIP: OnceLock<Vec<bool>> = OnceLock::new();
static FREQ_CACHE_FTABLES: OnceLock<FTables> = OnceLock::new();
// FrequencyCacheEntry and FrequencyCacheValue are structs for --frequency-jsonl JSON cache
#[derive(Serialize, Deserialize)]
struct FrequencyCacheEntry {
    field:       String,
    cardinality: u64,
    frequencies: Vec<FrequencyCacheValue>,
}

#[derive(Serialize, Deserialize)]
struct FrequencyCacheValue {
    value:      String,
    count:      u64,
    percentage: f64,
}

// FrequencyCacheMetadata stores how the cache was generated.
// Written as the first line of a JSONL cache file (.freq.csv.data.jsonl),
// followed by one FrequencyCacheEntry per subsequent line.
// Similar to StatsArgs in stats.rs, the metadata fields enable cache validation.
#[derive(Serialize, Deserialize)]
struct FrequencyCacheMetadata {
    arg_input:                String,
    flag_high_card_threshold: usize,
    flag_high_card_pct:       u8,
    flag_no_nulls:            bool,
    flag_no_headers:          bool,
    flag_delimiter:           String,
    record_count:             u64,
    column_count:             usize,
    date_generated:           String,
    qsv_version:              String,
    /// Selection signature: a unit-separator-joined string of the selected
    /// header bytes (in selection order). Lets us refuse cache reuse when
    /// the user runs frequency with a different `--select` (or different
    /// column ordering) than the one that produced the cache. Especially
    /// important in `--no-headers` mode, where cache entries are keyed by
    /// position-within-selection (`"1"`, `"2"`, ...) and would otherwise
    /// silently match a differently-ordered selection.
    /// `#[serde(default)]` keeps caches written before this field was added
    /// readable; an empty signature triggers a one-time fallback in
    /// `try_output_from_cache`.
    #[serde(default)]
    selection_signature:      String,
    /// Canonical (symlink-resolved, absolute) path of the source CSV at the time
    /// the cache was generated. Lets tooling (e.g. `qsv clean --stale`) resolve the
    /// exact source — including its real extension — instead of guessing from the
    /// cache stem. `#[serde(default)]` keeps pre-existing caches (without this
    /// field) readable; an empty value means "unknown source" and callers should
    /// treat such legacy caches conservatively.
    #[serde(default)]
    canonical_input_path:     String,
}

/// A neutral, read-only view of a frequency cache for consumers outside the
/// `frequency` command (currently `viz smart`) that want each column's top
/// value/count pairs without re-scanning the data.
pub(crate) struct FreqCacheView {
    /// Map of column field name -> per-column data. For `--no-headers` caches the keys are
    /// 1-based positional strings ("1", "2", ...). The `<HIGH_CARDINALITY>` / `<ALL_UNIQUE>`
    /// sentinel columns are omitted, so a missing key signals "recompute this column".
    pub columns: std::collections::HashMap<String, FreqCacheColumn>,
}

/// One column's data from a frequency cache, split into the non-null `(value, count)` pairs
/// and the count of empty/null cells (the cache's empty-string bucket). Kept separate so a
/// consumer can render the null bucket as its own bar (or drop it) rather than mixing it into
/// the value list.
pub(crate) struct FreqCacheColumn {
    /// Non-null `(value, count)` pairs, in the cache's stored order (count descending).
    pub pairs:      Vec<(String, u64)>,
    /// Count of empty/null cells for this column (0 if the cache had no null bucket).
    pub null_count: u64,
}

/// The frequency JSONL cache path for `path`, for consumers outside the `frequency` command
/// (currently `viz smart --dict-info`, which offers the cache it actually read as a download).
/// Delegates to `Args::cache_path_for` so the path is derived EXACTLY as
/// `read_frequency_cache_view` derives it — including its internal canonicalization, which is what
/// makes `data.csv`, `./data.csv` and a symlink to it all resolve to the same cache file.
pub(crate) fn frequency_cache_path(path: &std::path::Path) -> std::path::PathBuf {
    Args::cache_path_for(path)
}

/// Read and validate a frequency JSONL cache for `path`, independent of any
/// `frequency::Args`, for read-only reuse by other commands. Returns `None`
/// when the cache is absent, stale (older than the source), or was generated
/// with options incompatible with `(no_nulls, no_headers, delimiter)`.
///
/// Unlike `Args::read_frequency_cache`, this does NO `--select` filtering — it
/// returns the full set of cached columns and the caller looks up the ones it
/// needs by name. It DOES, however, reject caches that would be ambiguous for a
/// whole-file, name-keyed reader:
/// - In `--no-headers` mode, cache columns are keyed positionally within the selection that built
///   them, so a cache made with a non-identity `--select` (or with a legacy empty signature) would
///   alias the wrong columns. The cache is only trusted when its column count equals the full
///   input's AND the recorded selection signature matches the full input in original order AND that
///   signature is unambiguous: all first-row values distinct, all valid UTF-8 (the signature
///   stringifies fields lossily, so two distinct invalid-UTF8 values could collapse to the same
///   text), and none containing the `\x1f` separator the signature joins on (the join is
///   unescaped). Equal values, lossy collapse, or an embedded separator could otherwise mask a
///   reordering.
/// - Duplicate column names (across ALL entries, including sentinels) would make a name-keyed
///   lookup return the wrong column's data, so such caches are refused.
pub(crate) fn read_frequency_cache_view(
    path: &std::path::Path,
    no_nulls: bool,
    no_headers: bool,
    delimiter: Option<Delimiter>,
) -> Option<FreqCacheView> {
    use filetime::FileTime;

    let cache_path = Args::cache_path_for(path);
    if !cache_path.exists() {
        return None;
    }

    // Cache must be newer than the source CSV, else it's stale.
    let csv_metadata = fs::metadata(path).ok()?;
    let cache_fs_metadata = fs::metadata(&cache_path).ok()?;
    if FileTime::from_last_modification_time(&cache_fs_metadata)
        <= FileTime::from_last_modification_time(&csv_metadata)
    {
        log::info!("Frequency cache is stale; recomputing bar frequencies.");
        return None;
    }

    // JSONL: line 1 = metadata, remaining lines = per-column entries.
    let jsonl_content = fs::read_to_string(&cache_path).ok()?;
    let mut lines = jsonl_content.lines();
    let metadata: FrequencyCacheMetadata = serde_json::from_str(lines.next()?).ok()?;

    // Option compatibility — these affect what's stored / how columns are named
    // / how values were split, so a mismatch means the cache can't be reused.
    let current_delimiter =
        delimiter.map_or_else(|| ",".to_string(), |d| (d.as_byte() as char).to_string());
    if metadata.flag_no_nulls != no_nulls
        || metadata.flag_no_headers != no_headers
        || metadata.flag_delimiter != current_delimiter
    {
        log::info!("Frequency cache incompatible with current options; recomputing.");
        return None;
    }

    // In `--no-headers` mode the cache keys columns positionally within the
    // selection that produced it ("1", "2", ...). A caller reading the whole
    // file in original order would mis-map those keys if the cache was built
    // with a reordered/subset `--select`. The only identifier the cache records
    // for the selection is `selection_signature` — the first-row bytes joined in
    // selection order with a `\x1f` (Unit Separator) delimiter. That can PROVE
    // original order only when (a) the cache covers exactly all columns, (b) the
    // signature matches the full input in original order, (c) the first-row
    // values are all distinct (equal values could let a reordered selection
    // produce an identical signature), and (d) NO first-row value itself
    // contains the `\x1f` separator — otherwise the unescaped join is ambiguous
    // and a reordering could still collide even with distinct values. Reject
    // conservatively when any of these can't be established. (Headed caches are
    // keyed by name, so they self-protect and need no signature check.)
    if no_headers {
        // must match the separator used by `Args::selection_signature`
        const SIG_SEPARATOR: u8 = 0x1f;

        let path_str = path.to_string_lossy().into_owned();
        let rconfig = Config::new(Some(&path_str))
            .delimiter(delimiter)
            .no_headers_flag(true);
        let mut rdr = rconfig.reader().ok()?;
        let full_headers = rdr.byte_headers().ok()?.clone();

        // `selection_signature` stringifies each field with a LOSSY UTF-8
        // conversion and joins on `\x1f` without escaping. For the signature to
        // unambiguously prove identity column order we need every first-row value
        // to be (1) valid UTF-8 — so the lossy conversion is faithful and two
        // distinct invalid-UTF8 fields can't collapse to the same replacement
        // text — and (2) free of the `\x1f` separator — so the join splits back
        // to the original fields. Given those, distinct raw bytes ⇒ distinct
        // signature positions ⇒ a matching signature can only be the identity
        // permutation.
        let mut seen_vals: HashSet<Vec<u8>> = HashSet::new();
        let all_distinct = full_headers.iter().all(|f| seen_vals.insert(f.to_vec()));
        let all_utf8 = full_headers
            .iter()
            .all(|f| simdutf8::basic::from_utf8(f).is_ok());
        let has_separator = full_headers.iter().any(|f| f.contains(&SIG_SEPARATOR));
        let expected_sig = Args::selection_signature(&full_headers);

        if !all_distinct
            || !all_utf8
            || has_separator
            || metadata.column_count != full_headers.len()
            || metadata.selection_signature.is_empty()
            || metadata.selection_signature != expected_sig
        {
            log::info!(
                "Frequency cache selection can't be proven to match the full input \
                 (--no-headers); recomputing."
            );
            return None;
        }
    }

    let mut columns: std::collections::HashMap<String, FreqCacheColumn> =
        std::collections::HashMap::new();
    // Track every column name seen (sentinels included). A duplicate name makes a
    // name-keyed lookup ambiguous regardless of whether one duplicate is a
    // sentinel, so detect it before the sentinel skip below and refuse the cache.
    let mut seen_fields: HashSet<String> = HashSet::new();
    for line in lines {
        if line.is_empty() {
            continue;
        }
        let entry: FrequencyCacheEntry = serde_json::from_str(line).ok()?;
        if !seen_fields.insert(entry.field.clone()) {
            log::info!(
                "Frequency cache has duplicate column names; recomputing to avoid ambiguity."
            );
            return None;
        }
        // Skip sentinel columns (no usable per-value data) — leaving them out of
        // the map makes the caller fall back to recomputation for that column.
        if entry.frequencies.len() == 1
            && matches!(
                entry.frequencies[0].value.as_str(),
                "<HIGH_CARDINALITY>" | "<ALL_UNIQUE>"
            )
        {
            continue;
        }
        // Split the column into its non-null pairs and the empty/null bucket count, so the
        // consumer can render the null bucket as its own bar (or drop it) rather than mixing
        // it into the value list.
        let mut null_count: u64 = 0;
        let pairs: Vec<(String, u64)> = entry
            .frequencies
            .into_iter()
            .filter_map(|f| {
                if f.value.is_empty() {
                    null_count += f.count;
                    None
                } else {
                    Some((f.value, f.count))
                }
            })
            .collect();
        columns.insert(entry.field, FreqCacheColumn { pairs, null_count });
    }
    if columns.is_empty() {
        return None;
    }

    Some(FreqCacheView { columns })
}

// FrequencyEntry, FrequencyField and FrequencyOutput are
// structs for JSON output
#[derive(Serialize)]
struct FrequencyEntry {
    value:      String,
    count:      u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    percentage: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    rank:       Option<f64>,
}

#[derive(Serialize)]
struct FrequencyField {
    field:            String,
    r#type:           String,
    cardinality:      u64,
    nullcount:        u64,
    sparsity:         f64,
    uniqueness_ratio: f64,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    stats:            Vec<FieldStats>,
    frequencies:      Vec<FrequencyEntry>,
}

#[derive(Serialize, Clone)]
struct FieldStats {
    name:  String,
    value: JsonValue,
}

#[derive(Serialize)]
struct FrequencyOutput {
    input:         String,
    description:   String,
    rowcount:      u64,
    fieldcount:    usize,
    fields:        Vec<FrequencyField>,
    rank_strategy: RankStrategy,
}

// Shared frequency processing result
// used by both CSV and JSON output
#[derive(Clone)]
struct ProcessedFrequency {
    count:                u64,
    percentage:           f64,
    formatted_percentage: String,
    value:                Vec<u8>,
    rank:                 f64,
}

/// Estimates memory usage per record for frequency table computation.
///
/// This function calculates the approximate memory footprint of a single CSV record
/// when computing frequency tables. The estimate includes:
/// - Base record size (sum of field lengths)
/// - Hashmap overhead for storing unique values in Frequencies<Vec<u8>>
///
/// # Arguments
///
/// * `record` - The CSV record to estimate memory for
///
/// # Returns
///
/// Estimated memory in bytes per record
fn estimate_record_memory_for_frequency(record: &csv::ByteRecord) -> usize {
    // Base memory: sum of all field lengths
    let base_size: usize = record.iter().map(<[u8]>::len).sum();

    // Hashmap overhead: Frequencies<Vec<u8>> stores unique values
    // Each hashmap entry has overhead (~24 bytes) plus the value size
    // We estimate based on average field size
    let avg_field_size = if record.is_empty() {
        0
    } else {
        base_size / record.len()
    };

    // Estimate hashmap overhead: ~24 bytes per entry + value size
    // For frequency tables, we store each unique value once
    // Conservative estimate: assume we'll store all field values
    let hashmap_overhead = record.len() * (24 + avg_field_size);

    // Add overhead for Vec capacity
    let overhead = base_size / 4;

    base_size + hashmap_overhead + overhead
}

/// Helper function to calculate average record size from samples
fn calculate_avg_record_size_for_frequency(samples: &[csv::ByteRecord]) -> usize {
    if samples.is_empty() {
        1024 // Default: 1KB per record
    } else {
        let total_size: usize = samples
            .iter()
            .map(estimate_record_memory_for_frequency)
            .sum();
        (total_size / samples.len()).max(1024)
    }
}

/// Estimates total memory required for processing a chunk of records for frequency tables.
///
/// # Arguments
///
/// * `record_count` - Number of records in the chunk
/// * `avg_record_size` - Average size of a record in bytes
/// * `field_count` - Number of fields in the record
///
/// # Returns
///
/// Estimated total memory in bytes for the chunk
const fn estimate_chunk_memory_for_frequency(
    record_count: usize,
    avg_record_size: usize,
    field_count: usize,
) -> usize {
    // Base memory for records
    let base_memory = record_count.saturating_mul(avg_record_size);

    // Hashmap overhead: frequency tables store unique values
    // Estimate based on cardinality (assume 10% of records are unique per field)
    // Each hashmap entry: ~24 bytes overhead + value size
    let estimated_unique_per_field = if record_count / 10 > 0 {
        record_count / 10
    } else {
        1
    };
    let field_count_divisor = if field_count > 0 { field_count } else { 1 };
    let hashmap_overhead = estimated_unique_per_field
        .saturating_mul(field_count)
        .saturating_mul(avg_record_size / field_count_divisor + 24);

    // Add overhead for data structures (Frequencies objects, Vec capacity, etc.)
    // Estimate 20% overhead
    let overhead = (base_memory + hashmap_overhead) / 5;

    base_memory
        .saturating_add(hashmap_overhead)
        .saturating_add(overhead)
}

/// Calculates memory-aware chunk size for parallel frequency table processing.
///
/// This function determines an appropriate chunk size based on:
/// - Available memory per chunk (if configured)
/// - Dynamic estimation via sampling (if `max_chunk_memory_mb` is Some(0))
/// - CPU-based chunking (fallback)
///
/// # Arguments
///
/// * `idx_count` - Total number of records in the file
/// * `njobs` - Number of parallel jobs
/// * `max_chunk_memory_mb` - Maximum memory per chunk in MB ( None = dynamic sizing based on
///   available memory, Some(0) = dynamic sizing based on sampling, Some(n) = fixed limit)
/// * `field_count` - Number of fields in the record
/// * `sample_records` - Optional slice of sample records for dynamic sizing
///
/// # Returns
///
/// Calculated chunk size (number of records per chunk)
fn calculate_memory_aware_chunk_size_for_frequency(
    idx_count: u64,
    njobs: usize,
    max_chunk_memory_mb: Option<u64>,
    sample_records: Option<&[csv::ByteRecord]>,
) -> usize {
    // Frequency always uses memory-aware chunking since it builds hash tables.
    // This function has three configuration paths:
    //   - None: dynamic sizing based on sample records and estimated memory usage.
    //   - Some(0): dynamic sizing, also based on sample records and estimated memory usage.
    //   - Some(n): fixed memory limit per chunk.
    // In all cases, chunk size is calculated using memory-based estimates, not CPU-based chunking.
    match max_chunk_memory_mb {
        None => {
            // No memory limit configured, use dynamic sizing
            util::calculate_dynamic_chunk_size(
                idx_count,
                njobs,
                sample_records,
                estimate_record_memory_for_frequency,
            )
        },
        Some(0) => {
            // Dynamic sizing: sample records to estimate average size
            util::calculate_dynamic_chunk_size(
                idx_count,
                njobs,
                sample_records,
                estimate_record_memory_for_frequency,
            )
        },
        Some(limit_mb) => {
            // Fixed memory limit per chunk
            #[allow(clippy::cast_precision_loss)]
            let max_memory_bytes = (limit_mb as usize * 1024 * 1024) as f64 * util::SAFETY_MARGIN;

            // Estimate average record size
            // If we can't estimate, use a conservative default (1KB per record)
            let avg_record_size = if let Some(samples) = sample_records {
                if samples.is_empty() {
                    1024 // Default: 1KB per record
                } else {
                    let total_size: usize = samples
                        .iter()
                        .map(estimate_record_memory_for_frequency)
                        .sum();
                    debug_assert!(total_size > 0, "total_size should be positive here");
                    (total_size / samples.len()).max(1024) // ensure minimum 1KB estimate, samples.len() is guaranteed to be positive here
                }
            } else {
                1024 // Default: 1KB per record
            };

            // Calculate chunk size based on memory limit
            #[allow(clippy::cast_precision_loss)]
            let chunk_size = (max_memory_bytes / (avg_record_size as f64).max(1.0)) as usize;

            // Ensure chunk size is reasonable
            chunk_size.max(1).min(idx_count as usize)
        },
    }
}

/// Check whether `--sketch-method frequent_items` can be auto-enabled given
/// the current flag configuration. Mirrors the explicit `frequent_items`
/// rejection block in `run()` (the `match args.flag_sketch_method` arm that
/// rejects each conflicting flag with `fail_incorrectusage_clierror!` and
/// validates `--sketch-map-size`) so an auto-enable only flips the method
/// when the user could have set it by hand without hitting any of those
/// errors.
///
/// `user_set_sketch_method` indicates whether the user passed `--sketch-method`
/// on the command line (regardless of value). When `true`, the auto-enable is
/// suppressed even if `flag_sketch_method == "exact"` — this honors the
/// "Re-run with explicit `--sketch-method exact` to disable the auto-enable"
/// contract surfaced in the wwarn and `--memcheck` docs. Without this guard,
/// the docopt default of `"exact"` would make an explicit `exact` indistinguishable
/// from omitting the flag, and the opt-out advice in the wwarn would be a no-op.
///
/// Returns `false` if any conflicting flag is set, if the user explicitly set
/// the method (regardless of value), or if --sketch-map-size is invalid.
///
/// On big-endian targets the Apache `DataSketches` port is unavailable, so this
/// function compiles to a stub that always returns `false`. The OOM auto-enable
/// path then leaves `flag_sketch_method` alone and the error propagates.
#[cfg(not(target_endian = "big"))]
fn can_enable_frequent_items(args: &Args, user_set_sketch_method: bool) -> bool {
    if args.flag_sketch_method != "exact" || user_set_sketch_method {
        return false;
    }
    if args.flag_asc
        || args.flag_weight.is_some()
        || args.flag_ignore_case
        || args.flag_no_trim
        || args.flag_other_sorted
        || args.flag_null_sorted
        || args.flag_frequency_jsonl
        || args.flag_json
        || args.flag_pretty_json
        || args.flag_toon
    {
        return false;
    }
    #[cfg(feature = "luau")]
    if args.flag_stats_filter.is_some() {
        return false;
    }
    // sketch-map-size validity (mirrors the `must be a power of two and >= 8`
    // check in run()'s frequent_items dispatch).
    args.flag_sketch_map_size >= 8 && args.flag_sketch_map_size.is_power_of_two()
}

/// Big-endian stub: Apache DataSketches is unavailable on big-endian targets,
/// so the Frequent Items sketch cannot be auto-enabled. The OOM path falls
/// through to returning the original error.
#[cfg(target_endian = "big")]
fn can_enable_frequent_items(_args: &Args, _user_set_sketch_method: bool) -> bool {
    false
}

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

    // Detect whether the user explicitly passed --sketch-method on the command
    // line. docopt fills in the default value ("exact") regardless, so without
    // this scan we can't honor an explicit `--sketch-method exact` opt-out
    // during the OOM auto-fallback.
    let user_set_sketch_method = util::argv_has_flag(argv, "--sketch-method");

    // Handle --no-other flag (alias for --other-text "<NONE>")
    if args.flag_no_other {
        args.flag_other_text = "<NONE>".to_string();
    }

    // Handle --null-text "<NONE>" (alias for --no-nulls)
    if args.flag_null_text == "<NONE>" {
        args.flag_no_nulls = true;
    }

    let mut rconfig = args.rconfig();

    let is_stdin = rconfig.is_stdin();

    // Validate --frequency-jsonl early, before any computation
    if args.flag_frequency_jsonl && is_stdin {
        return fail_clierror!("--frequency-jsonl requires a file input, not stdin.");
    }

    // if stdin and args.flag_json is true, save stdin to tempfile
    // so we can derive stats
    let mut stdin_temp_file;
    let is_json = args.flag_json || args.flag_pretty_json || args.flag_toon;
    if is_stdin && is_json {
        let temp_dir = std::env::temp_dir();
        stdin_temp_file = tempfile::Builder::new()
            .suffix(".csv")
            .tempfile_in(&temp_dir)?;
        io::copy(&mut io::stdin(), &mut stdin_temp_file)?;
        args.arg_input = Some(stdin_temp_file.path().to_string_lossy().to_string());
        rconfig = args.rconfig();
    }

    // --sketch-method: dispatch to the Frequent Items path when requested. The FI
    // path is intentionally narrow — many existing flags don't compose with a
    // streaming sketch that only tracks heavy hitters. Reject the unsupported
    // combinations early so users get a clear error instead of silently-wrong
    // output.
    args.flag_sketch_method = args.flag_sketch_method.to_lowercase();
    match args.flag_sketch_method.as_str() {
        "exact" => {},
        // Apache DataSketches is unavailable on big-endian targets, so the
        // Frequent Items sketch cannot run there. Reject upfront with a clear
        // platform message instead of falling through to the normal validation.
        #[cfg(target_endian = "big")]
        "frequent_items" => {
            return fail_incorrectusage_clierror!(
                "--sketch-method frequent_items requires a little-endian target. Apache \
                 DataSketches is not available on big-endian platforms (e.g., s390x). Use \
                 --sketch-method exact."
            );
        },
        #[cfg(not(target_endian = "big"))]
        "frequent_items" => {
            if args.flag_asc {
                return fail_incorrectusage_clierror!(
                    "--sketch-method frequent_items tracks heavy hitters only — least-frequent \
                     items are not recoverable. Remove --asc or use --sketch-method exact."
                );
            }
            if args.flag_weight.is_some() {
                return fail_incorrectusage_clierror!(
                    "--sketch-method frequent_items operates on unit-weight streams. Use \
                     --sketch-method exact for weighted frequencies."
                );
            }
            if args.flag_ignore_case {
                return fail_incorrectusage_clierror!(
                    "--sketch-method frequent_items does not support --ignore-case. Use \
                     --sketch-method exact."
                );
            }
            if args.flag_no_trim {
                return fail_incorrectusage_clierror!(
                    "--sketch-method frequent_items does not support --no-trim. Use \
                     --sketch-method exact."
                );
            }
            if args.flag_other_sorted {
                return fail_incorrectusage_clierror!(
                    "--sketch-method frequent_items does not support --other-sorted. The sketch \
                     always emits the Other row at the end. Use --sketch-method exact."
                );
            }
            if args.flag_null_sorted {
                return fail_incorrectusage_clierror!(
                    "--sketch-method frequent_items does not support --null-sorted. The sketch \
                     ranks NULL alongside other values by estimate. Use --sketch-method exact."
                );
            }
            if args.flag_frequency_jsonl {
                return fail_incorrectusage_clierror!(
                    "--sketch-method frequent_items is not supported with --frequency-jsonl."
                );
            }
            if is_json {
                return fail_incorrectusage_clierror!(
                    "--sketch-method frequent_items does not support --json/--pretty-json/--toon \
                     output. Use --sketch-method exact."
                );
            }
            #[cfg(feature = "luau")]
            if args.flag_stats_filter.is_some() {
                return fail_incorrectusage_clierror!(
                    "--sketch-method frequent_items is not supported with --stats-filter."
                );
            }
            let n = args.flag_sketch_map_size;
            if n < 8 || !n.is_power_of_two() {
                return fail_incorrectusage_clierror!(
                    "--sketch-map-size must be a power of two and >= 8, got {n}."
                );
            }
            return args.run_frequent_items(&rconfig);
        },
        other => {
            return fail_incorrectusage_clierror!(
                "Invalid --sketch-method: {other}. Choose 'exact' or 'frequent_items'."
            );
        },
    }

    // Check if we have an index and will use parallel processing
    // If so, skip mem_file_check since memory-aware chunking will handle it
    // Reuse the existing `rconfig` (not a fresh `args.rconfig()`): for special-format
    // inputs (e.g. `.zip`) each `Config` lazily resolves to its own temp path, so a
    // fresh Config would check an index next to a *different* temp than the one used
    // for `resolved_path()`/auto-indexing below.
    let mut indexed_result = rconfig.indexed()?;
    let will_use_parallel = match &indexed_result {
        Some(_) => {
            // We have an index, check if we'll use parallel processing
            match args.flag_jobs {
                Some(num_jobs) => num_jobs != 1,
                _ => true, // Default to parallel when index exists
            }
        },
        None => false, // No index, will use sequential
    };

    // we're loading the entire file into memory, we need to check avail mem
    // Skip this check for parallel processing since memory-aware chunking handles it
    if !will_use_parallel && let Some(path) = rconfig.resolved_path()? {
        // Try mem_file_check, and if it fails for an unindexed file, auto-create index
        match util::mem_file_check(&path, false, args.flag_memcheck) {
            Ok(_) => {
                // Memory check passed, proceed with sequential processing
            },
            Err(e) => {
                // Memory check failed. We have two fallbacks that can stack:
                //   1. auto-create an index to switch to parallel/indexed mode
                //   2. auto-enable the Frequent Items sketch (Misra-Gries) to bound per-column
                //      memory regardless of sequential vs. parallel
                // Only propagate the original OOM error if NEITHER fallback engages.
                let mut index_succeeded = false;
                if indexed_result.is_none() && !rconfig.is_stdin() {
                    log::info!(
                        "File too large for sequential processing. Auto-creating index to enable \
                         parallel processing..."
                    );
                    match util::create_index_for_file(&path, &rconfig) {
                        Ok(()) => {
                            indexed_result = rconfig.indexed()?;
                            index_succeeded = indexed_result.is_some();
                            if index_succeeded {
                                log::info!(
                                    "Index created successfully. Switching to parallel processing."
                                );
                            }
                        },
                        Err(index_err) => {
                            log::warn!("Failed to auto-create index: {index_err}");
                        },
                    }
                }

                // Sketch fallback: only for OOM (not other CliErrors).
                // The Frequent Items dispatch at frequency.rs:787 normally fires before
                // mem_file_check runs, so we re-route into run_frequent_items() directly
                // here rather than mutating flag_sketch_method and falling through —
                // run_frequent_items doesn't consult flag_sketch_method, so the field
                // stays at "exact" and that's fine (cache key / diagnostics use other
                // paths). Verified via grep over run_frequent_items's body.
                //
                // Apache DataSketches is unavailable on big-endian targets, so the sketch
                // fallback is gated out entirely there. The big-endian branch below
                // preserves the "no index → propagate the error" behavior.
                #[cfg(not(target_endian = "big"))]
                if matches!(e, crate::CliError::OutOfMemory(_))
                    && can_enable_frequent_items(&args, user_set_sketch_method)
                {
                    wwarn!(
                        "OOM during memory check: auto-enabling --sketch-method frequent_items \
                         (Misra-Gries, map size {n}). Re-run with explicit --sketch-method exact \
                         to disable the auto-enable.",
                        n = args.flag_sketch_map_size,
                    );
                    return args.run_frequent_items(&rconfig);
                } else if !index_succeeded {
                    return Err(e);
                }
                // Big-endian: no sketch auto-enable is possible (DataSketches is
                // compiled out), so the only recovery left is the index that may
                // have been created above. If indexing did NOT succeed, propagate
                // the original error. If it DID succeed, silently fall through and
                // retry with parallel processing — matching the little-endian
                // behavior on the same branch (the original `else if !index_succeeded`
                // arm above) where success similarly swallows `e`.
                #[cfg(target_endian = "big")]
                if !index_succeeded {
                    return Err(e);
                }
            },
        }
    }

    // Create NULL_VAL, ALL_UNIQUE_TEXT & HIGH_CARDINALITY_TEXT once at the start to avoid
    // repeated string & vec allocations in hot loops.
    // safety: we're initializing the OnceLocks at the start of the program
    NULL_VAL
        .set(args.flag_null_text.as_bytes().to_vec())
        .unwrap();

    ALL_UNIQUE_TEXT
        .set(args.flag_all_unique_text.as_bytes().to_vec())
        .unwrap();

    if args.flag_frequency_jsonl {
        // Guard: --frequency-jsonl is incompatible with computation-changing flags
        if args.flag_ignore_case {
            return fail_incorrectusage_clierror!(
                "--frequency-jsonl cannot be used with --ignore-case."
            );
        }
        if args.flag_no_trim {
            return fail_incorrectusage_clierror!(
                "--frequency-jsonl cannot be used with --no-trim."
            );
        }
        if args.flag_weight.is_some() {
            return fail_incorrectusage_clierror!(
                "--frequency-jsonl cannot be used with --weight."
            );
        }

        // Override high-card thresholds from env vars when CLI defaults are used
        if args.flag_high_card_threshold == 100
            && let Ok(val) = std::env::var("QSV_FREQ_HIGH_CARD_THRESHOLD")
            && let Ok(parsed) = val.parse::<usize>()
        {
            args.flag_high_card_threshold = parsed;
        }
        if args.flag_high_card_pct == 90
            && let Ok(val) = std::env::var("QSV_FREQ_HIGH_CARD_PCT")
            && let Ok(parsed) = val.parse::<u8>()
        {
            args.flag_high_card_pct = parsed;
        }

        // Validate --high-card-pct is in range 1-100
        if args.flag_high_card_pct == 0 || args.flag_high_card_pct > 100 {
            return fail_incorrectusage_clierror!(
                "--high-card-pct must be between 1 and 100, got {}.",
                args.flag_high_card_pct
            );
        }
    }

    // Try to use frequency cache before computing FTables.
    // Cache is eligible when: not stdin, not JSON output, not --force,
    // and no computation-changing flags (--ignore-case, --no-trim, --weight).
    // Also skip when --no-float or --stats-filter are active (these need stats cache lookups).
    #[cfg(feature = "luau")]
    let has_stats_filter = args.flag_stats_filter.is_some();
    #[cfg(not(feature = "luau"))]
    let has_stats_filter = false;

    let can_use_freq_cache = !is_stdin
        && !is_json
        && !args.flag_force
        && !args.flag_frequency_jsonl
        && !args.flag_ignore_case
        && !args.flag_no_trim
        && args.flag_weight.is_none()
        && args.flag_no_float.is_none()
        && !has_stats_filter;

    if can_use_freq_cache && args.try_output_from_cache(&rconfig, is_json)? {
        return Ok(());
    }

    let (headers, mut tables, weighted_tables) = if let Some(idx) = indexed_result
        && util::njobs(args.flag_jobs) > 1
    {
        args.parallel_ftables(&idx, &rconfig)
    } else {
        args.sequential_ftables()
    }?;

    // Merge cached frequency tables for columns that were skipped during computation.
    // In the partial cache path, some columns have pre-built FTables from the frequency
    // cache (non-HIGH_CARDINALITY), while HIGH_CARDINALITY columns were computed above.
    if let (Some(cache_skip), Some(cached_ftables)) =
        (FREQ_CACHE_SKIP.get(), FREQ_CACHE_FTABLES.get())
    {
        debug_assert_eq!(
            cache_skip.len(),
            tables.len(),
            "FREQ_CACHE_SKIP length ({}) != tables length ({})",
            cache_skip.len(),
            tables.len()
        );
        debug_assert_eq!(
            cached_ftables.len(),
            tables.len(),
            "FREQ_CACHE_FTABLES length ({}) != tables length ({})",
            cached_ftables.len(),
            tables.len()
        );
        for (i, &is_cached) in cache_skip.iter().enumerate() {
            if is_cached && i < tables.len() && i < cached_ftables.len() {
                tables[i] = cached_ftables[i].clone();
            }
        }
    }

    // Write frequency cache if --frequency-jsonl is set.
    // Always write when explicitly requested — the user may be regenerating
    // with different thresholds or after data changes.
    if args.flag_frequency_jsonl {
        args.write_frequency_jsonl(&headers, &tables, &rconfig)?;
    }

    if is_json {
        return args.output_json(
            &headers,
            tables,
            weighted_tables.as_ref(),
            &rconfig,
            argv,
            is_stdin,
        );
    }

    let row_count = *FREQ_ROW_COUNT.get().unwrap_or(&0);
    let abs_dec_places = args.flag_pct_dec_places.unsigned_abs() as u32;

    // safety: we know that UNIQUE_COLUMNS has been previously set
    // when compiling frequencies by sel_headers fn in either sequential or parallel mode
    let unique_headers_vec = UNIQUE_COLUMNS_VEC.get().unwrap();

    let mut wtr = Config::new(args.flag_output.as_ref()).writer()?;
    // write headers
    wtr.write_record(vec!["field", "value", "count", "percentage", "rank"])?;

    // Handle weighted vs unweighted frequencies
    if let Some(ref weighted) = weighted_tables {
        // Process weighted frequencies — the inline loop differs from the
        // unweighted helper because process_frequencies_weighted takes the
        // weighted column map by reference, not the FTable.
        let mut header_vec: Vec<u8>;
        let mut itoa_buffer = itoa::Buffer::new();
        let mut zmij_buffer = zmij::Buffer::new();
        let mut rank_buffer = String::with_capacity(20);
        #[allow(unused_assignments)]
        let mut value_str = String::with_capacity(100);
        let mut processed_frequencies: Vec<ProcessedFrequency> = Vec::with_capacity(headers.len());
        let vis_whitespace = args.flag_vis_whitespace;

        for (i, header) in headers.iter().enumerate() {
            header_vec = if rconfig.no_headers {
                (i + 1).to_string().into_bytes()
            } else {
                header.to_vec()
            };

            if i < weighted.len() {
                args.process_frequencies_weighted(
                    unique_headers_vec.contains(&i),
                    abs_dec_places,
                    row_count,
                    &weighted[i],
                    &mut processed_frequencies,
                );
            }

            for processed_freq in &processed_frequencies {
                // Format rank: show as integer if whole number, otherwise with decimals
                // Sentinel value -1.0 indicates NULL entry with --pct-nulls=false (empty rank)
                rank_buffer.clear();
                if processed_freq.rank >= 0.0 {
                    if processed_freq.rank.fract() == 0.0 {
                        rank_buffer.push_str(itoa_buffer.format(processed_freq.rank as u64));
                    } else {
                        rank_buffer.push_str(zmij_buffer.format(processed_freq.rank));
                    }
                }

                let value_bytes: &[u8] = if vis_whitespace {
                    value_str =
                        util::visualize_whitespace(&util::bytes_to_cow_str(&processed_freq.value));
                    value_str.as_bytes()
                } else {
                    &processed_freq.value
                };

                wtr.write_record([
                    &*header_vec,
                    value_bytes,
                    itoa_buffer.format(processed_freq.count).as_bytes(),
                    processed_freq.formatted_percentage.as_bytes(),
                    rank_buffer.as_bytes(),
                ])?;
            }
            processed_frequencies.clear();
        }
    } else {
        args.emit_unweighted_csv_rows(
            &mut wtr,
            &headers,
            tables,
            unique_headers_vec,
            abs_dec_places,
            row_count,
            rconfig.no_headers,
        )?;
    }
    Ok(wtr.flush()?)
}

type Headers = csv::ByteRecord;
type FTable = Frequencies<Vec<u8>>;
type FTables = Vec<Frequencies<Vec<u8>>>;
// Weighted frequency tables: HashMap for each column storing value -> weighted count
type WeightedFTables = Vec<HashMap<Vec<u8>, f64>>;

// Below `MERGE_PARALLEL_THRESHOLD` columns, the rayon overhead exceeds the
// per-column merge cost, so we fall back to a sequential zip.
const MERGE_PARALLEL_THRESHOLD: usize = 8;

// Pairwise merge of two FTables. Empty acts as identity so this composes with
// rayon `reduce(Vec::new, ...)`. Per-column Frequencies merges run in parallel
// across the rayon pool when there are enough columns to amortize the overhead.
fn merge_ftables(mut a: FTables, b: FTables) -> FTables {
    if a.is_empty() {
        return b;
    }
    if b.is_empty() {
        return a;
    }
    // Real assert (not debug_assert) because `zip` would silently truncate
    // and drop columns on mismatch. The invariant is structural (all chunks
    // share `sel`), but enforcing it in release prevents silent corruption
    // if that ever changes.
    assert_eq!(
        a.len(),
        b.len(),
        "all chunks share the same column selection"
    );
    if a.len() < MERGE_PARALLEL_THRESHOLD {
        for (left, right) in a.iter_mut().zip(b) {
            left.merge(right);
        }
    } else {
        a.par_iter_mut()
            .zip(b.into_par_iter())
            .for_each(|(left, right)| left.merge(right));
    }
    a
}

// Pairwise merge of two WeightedFTables — per-column HashMap fold, parallel over columns
// once column count clears `MERGE_PARALLEL_THRESHOLD`.
fn merge_weighted_ftables(mut a: WeightedFTables, b: WeightedFTables) -> WeightedFTables {
    if a.is_empty() {
        return b;
    }
    if b.is_empty() {
        return a;
    }
    // Real assert (see merge_ftables for rationale).
    assert_eq!(
        a.len(),
        b.len(),
        "all chunks share the same column selection"
    );
    let merge_pair = |left: &mut HashMap<Vec<u8>, f64>, right: HashMap<Vec<u8>, f64>| {
        left.reserve(right.len());
        for (k, v) in right {
            *left.entry(k).or_insert(0.0) += v;
        }
    };
    if a.len() < MERGE_PARALLEL_THRESHOLD {
        for (left, right) in a.iter_mut().zip(b) {
            merge_pair(left, right);
        }
    } else {
        a.par_iter_mut()
            .zip(b.into_par_iter())
            .for_each(|(left, right)| merge_pair(left, right));
    }
    a
}

#[allow(clippy::cast_precision_loss)]
fn apply_ranking_strategy_unweighted(
    groups: Vec<(u64, Vec<Vec<u8>>)>,
    strategy: RankStrategy,
    pct_factor: f64,
    null_val: &[u8],
    pct_nulls: bool,
) -> (Vec<(Vec<u8>, u64, f64, f64)>, u64, f64) {
    let mut counts_final: Vec<(Vec<u8>, u64, f64, f64)> =
        Vec::with_capacity(groups.iter().map(|(_, group)| group.len()).sum::<usize>() + 1);
    let mut current_rank = 1.0_f64;
    let mut count_sum = 0_u64;
    let mut pct_sum = 0.0_f64;

    // Per-row emit shared by all five strategies. Returns true when the row
    // was a sentinel-suppressed null (pct_nulls=false + empty value), so
    // Ordinal can skip rank advancement to match Dense/Min/Max/Average
    // semantics for nulls.
    {
        let mut emit = |byte_string: Vec<u8>, count: u64, rank: f64| -> bool {
            count_sum += count;
            if byte_string.is_empty() {
                if pct_nulls {
                    let pct = count as f64 * pct_factor;
                    pct_sum += pct;
                    counts_final.push((null_val.to_vec(), count, pct, rank));
                    false
                } else {
                    counts_final.push((null_val.to_vec(), count, -1.0, -1.0));
                    true
                }
            } else {
                let pct = count as f64 * pct_factor;
                pct_sum += pct;
                counts_final.push((byte_string, count, pct, rank));
                false
            }
        };

        match strategy {
            RankStrategy::Dense => {
                // Dense ranking (1223)
                for (count, mut group) in groups {
                    group.sort_unstable();
                    for byte_string in group {
                        emit(byte_string, count, current_rank);
                    }
                    current_rank += 1.0;
                }
            },
            RankStrategy::Min => {
                // Standard competition ranking (1224)
                for (count, mut group) in groups {
                    group.sort_unstable();
                    let group_len = group.len();
                    for byte_string in group {
                        emit(byte_string, count, current_rank);
                    }
                    current_rank += group_len as f64;
                }
            },
            RankStrategy::Max => {
                // Modified competition ranking (1334)
                for (count, mut group) in groups {
                    group.sort_unstable();
                    let group_len = group.len();
                    let max_rank = current_rank + group_len as f64 - 1.0;
                    for byte_string in group {
                        emit(byte_string, count, max_rank);
                    }
                    current_rank += group_len as f64;
                }
            },
            RankStrategy::Ordinal => {
                // Ordinal ranking (1234). Sentinel-suppressed nulls do NOT
                // consume a rank slot, matching the other strategies.
                for (count, mut group) in groups {
                    group.sort_unstable();
                    for byte_string in group {
                        let suppressed = emit(byte_string, count, current_rank);
                        if !suppressed {
                            current_rank += 1.0;
                        }
                    }
                }
            },
            RankStrategy::Average => {
                // Fractional ranking (1 2.5 2.5 4)
                for (count, mut group) in groups {
                    group.sort_unstable();
                    let group_len = group.len();
                    let avg_rank = current_rank + (group_len as f64 - 1.0) / 2.0;
                    for byte_string in group {
                        emit(byte_string, count, avg_rank);
                    }
                    current_rank += group_len as f64;
                }
            },
        }
    }

    (counts_final, count_sum, pct_sum)
}

#[allow(clippy::cast_precision_loss)]
fn apply_ranking_strategy_weighted(
    groups: Vec<(f64, Vec<Vec<u8>>)>,
    strategy: RankStrategy,
    pct_factor: f64,
    null_val: &[u8],
    pct_nulls: bool,
) -> (Vec<(Vec<u8>, f64, f64, f64)>, f64, f64) {
    let mut counts_final: Vec<(Vec<u8>, f64, f64, f64)> =
        Vec::with_capacity(groups.iter().map(|(_, group)| group.len()).sum::<usize>() + 1);
    let mut current_rank = 1.0_f64;
    let mut count_sum = 0.0_f64;
    let mut pct_sum = 0.0_f64;

    // Per-row emit shared by all five strategies. Returns true when the row
    // was a sentinel-suppressed null (pct_nulls=false + empty value), so
    // Ordinal can skip rank advancement to match Dense/Min/Max/Average
    // semantics for nulls.
    {
        let mut emit = |byte_string: Vec<u8>, weight: f64, rank: f64| -> bool {
            count_sum += weight;
            if byte_string.is_empty() {
                if pct_nulls {
                    let pct = weight * pct_factor;
                    pct_sum += pct;
                    counts_final.push((null_val.to_vec(), weight, pct, rank));
                    false
                } else {
                    counts_final.push((null_val.to_vec(), weight, -1.0, -1.0));
                    true
                }
            } else {
                let pct = weight * pct_factor;
                pct_sum += pct;
                counts_final.push((byte_string, weight, pct, rank));
                false
            }
        };

        match strategy {
            RankStrategy::Dense => {
                // Dense ranking (1223)
                for (weight, mut group) in groups {
                    group.sort_unstable();
                    for byte_string in group {
                        emit(byte_string, weight, current_rank);
                    }
                    current_rank += 1.0;
                }
            },
            RankStrategy::Min => {
                // Standard competition ranking (1224)
                for (weight, mut group) in groups {
                    group.sort_unstable();
                    let group_len = group.len();
                    for byte_string in group {
                        emit(byte_string, weight, current_rank);
                    }
                    current_rank += group_len as f64;
                }
            },
            RankStrategy::Max => {
                // Modified competition ranking (1334)
                for (weight, mut group) in groups {
                    group.sort_unstable();
                    let group_len = group.len();
                    let max_rank = current_rank + group_len as f64 - 1.0;
                    for byte_string in group {
                        emit(byte_string, weight, max_rank);
                    }
                    current_rank += group_len as f64;
                }
            },
            RankStrategy::Ordinal => {
                // Ordinal ranking (1234). Sentinel-suppressed nulls do NOT
                // consume a rank slot, matching the other strategies.
                for (weight, mut group) in groups {
                    group.sort_unstable();
                    for byte_string in group {
                        let suppressed = emit(byte_string, weight, current_rank);
                        if !suppressed {
                            current_rank += 1.0;
                        }
                    }
                }
            },
            RankStrategy::Average => {
                // Fractional ranking (1 2.5 2.5 4)
                for (weight, mut group) in groups {
                    group.sort_unstable();
                    let group_len = group.len();
                    let avg_rank = current_rank + (group_len as f64 - 1.0) / 2.0;
                    for byte_string in group {
                        emit(byte_string, weight, avg_rank);
                    }
                    current_rank += group_len as f64;
                }
            },
        }
    }

    (counts_final, count_sum, pct_sum)
}

/// Apply limits to weighted frequency counts
///
/// # Arguments
/// * `counts` - Mutable reference to vector of `(value, weight)` pairs
/// * `limit` - Limit value; if positive, keep only the top N weighted values; if negative, keep
///   only values with weight greater than or equal to the absolute value of this limit; if zero, no
///   limits are applied
/// * `lmt_threshold` - Threshold controlling when limits are applied. Limits are applied when this
///   is 0 or when it is greater than or equal to the number of unique values; when this is a
///   positive number less than the unique count, no limits are applied.
fn apply_limits_weighted(counts: &mut Vec<(Vec<u8>, f64)>, limit: isize, lmt_threshold: usize) {
    let unique_counts_len = counts.len();
    if lmt_threshold == 0 || lmt_threshold >= unique_counts_len {
        let abs_limit = limit.unsigned_abs();

        #[allow(clippy::cast_precision_loss)]
        if limit > 0 {
            counts.truncate(abs_limit);
        } else if limit < 0 {
            let count_limit = abs_limit as f64;
            counts.retain(|(_, weight)| *weight >= count_limit);
        }
    }
}

/// Apply limits to unweighted frequency counts
///
/// # Arguments
/// * `counts` - Mutable reference to counts vector
/// * `limit` - Limit value (positive = top N, negative = threshold)
/// * `unq_limit` - Unique limit for all-unique columns
/// * `lmt_threshold` - Threshold controlling when limits are applied. Limits are applied when this
///   is 0 or when it is >= the number of unique values. When this is a positive number less than
///   the unique count, no limits are applied.
/// * `all_unique` - Whether the column has all unique values
fn apply_limits_unweighted(
    counts: &mut Vec<(Vec<u8>, u64)>,
    limit: isize,
    unq_limit: usize,
    lmt_threshold: usize,
    all_unique: bool,
) {
    let unique_counts_len = counts.len();
    if lmt_threshold == 0 || lmt_threshold >= unique_counts_len {
        let abs_limit = limit.unsigned_abs();
        let unique_limited = if all_unique && limit > 0 && unq_limit != abs_limit && unq_limit > 0 {
            counts.truncate(unq_limit);
            true
        } else {
            false
        };

        if limit > 0 {
            counts.truncate(abs_limit);
        } else if limit < 0 && !unique_limited {
            // if limit < 0, only return values with an occurrence count >= abs value of limit
            // Only do this if we haven't already unique limited the values
            let count_limit = abs_limit as u64;
            counts.retain(|(_, count)| *count >= count_limit);
        }
    }
}

/// Group unweighted frequency values by count.
///
/// # Arguments
/// * `counts` - A vector of `(value, count)` pairs, where `value` is the byte-string representation
///   of the category and `count` is its frequency.
///
/// # Returns
/// A vector of `(count, values)` pairs, where `values` is the list of
/// byte-strings that share the same `count`.
fn group_by_count(counts: Vec<(Vec<u8>, u64)>) -> Vec<(u64, Vec<Vec<u8>>)> {
    let mut count_groups: Vec<(u64, Vec<Vec<u8>>)> = Vec::new();
    let mut current_count: Option<u64> = None;
    let mut current_group = Vec::new();

    for (byte_string, count) in counts {
        if let Some(prev_count) = current_count
            && count != prev_count
        {
            count_groups.push((prev_count, std::mem::take(&mut current_group)));
        }

        current_count = Some(count);
        current_group.push(byte_string);
    }
    if let Some(prev_count) = current_count {
        count_groups.push((prev_count, current_group));
    }

    count_groups
}

/// Group weighted frequency values by weight (with tolerance).
///
/// # Arguments
///
/// * `counts` - A list of `(value, weight)` pairs, where `value` is a byte string and `weight` is
///   the numeric weight used for grouping. The vector is expected to be ordered by `weight` so that
///   equal (or near-equal) weights are adjacent.
/// * `tolerance` - The maximum absolute difference between consecutive weights for them to be
///   treated as belonging to the same group.
fn group_by_weight(counts: Vec<(Vec<u8>, f64)>, tolerance: f64) -> Vec<(f64, Vec<Vec<u8>>)> {
    let mut weight_groups: Vec<(f64, Vec<Vec<u8>>)> = Vec::new();
    let mut current_weight: Option<f64> = None;
    let mut current_group: Vec<Vec<u8>> = Vec::new();

    for (byte_string, weight) in counts {
        if let Some(prev_weight) = current_weight
            && (prev_weight - weight).abs() > tolerance
        {
            weight_groups.push((prev_weight, std::mem::take(&mut current_group)));
        }

        current_weight = Some(weight);
        current_group.push(byte_string);
    }
    if let Some(prev_weight) = current_weight {
        weight_groups.push((prev_weight, current_group));
    }

    weight_groups
}

/// Implementation of helper methods for frequency command arguments.
/// Provides configuration helpers and post-processing utilities for results.
impl Args {
    pub fn rconfig(&self) -> Config {
        Config::new(self.arg_input.as_ref())
            .delimiter(self.flag_delimiter)
            .no_headers_flag(self.flag_no_headers)
            .select(self.flag_select.clone())
    }

    /// Compute the cache path for a given input file, canonicalizing where
    /// possible so that `data.csv` and `./data.csv` (or symlinks) resolve to
    /// the same cache file. Falls back to the input path verbatim if
    /// canonicalization fails (e.g. permissions, exotic filesystems) — this
    /// preserves prior behavior and is logged at debug level.
    fn cache_path_for(path: &std::path::Path) -> std::path::PathBuf {
        let base = match fs::canonicalize(path) {
            Ok(p) => p,
            Err(e) => {
                log::debug!(
                    "Could not canonicalize input path {} for cache lookup: {e}. Falling back to \
                     uncanonicalized path.",
                    path.display()
                );
                path.to_path_buf()
            },
        };
        base.with_extension("freq.csv.data.jsonl")
    }

    /// Compute a stable signature for the selected headers (in selection
    /// order). Used to detect when a cache was generated with a different
    /// `--select` (different columns OR different ordering) than the current
    /// invocation. Especially important in `--no-headers` mode where cache
    /// entries are keyed by position-within-selection.
    ///
    /// The chosen separator (`\x1f`, ASCII Unit Separator) is highly unlikely
    /// to appear in real CSV column names or first-row data, so it
    /// unambiguously delimits fields without escaping.
    fn selection_signature(headers: &Headers) -> String {
        let cap: usize = headers.iter().map(|h| h.len() + 1).sum();
        let mut s = String::with_capacity(cap);
        for (i, field) in headers.iter().enumerate() {
            if i > 0 {
                s.push('\x1f');
            }
            s.push_str(&util::bytes_to_cow_str(field));
        }
        s
    }

    /// Write the complete frequency distribution as a JSON cache file
    /// (`.freq.csv.data.json`). The cache combines metadata (args,
    /// thresholds) and per-column data in a single JSON object.
    /// `ALL_UNIQUE` columns get a single `<ALL_UNIQUE>` sentinel entry.
    /// `HIGH_CARDINALITY` columns get a single `<HIGH_CARDINALITY>` sentinel entry.
    /// Normal columns get complete frequency data (all values, counts, percentages).
    #[allow(clippy::cast_precision_loss)]
    fn write_frequency_jsonl(
        &self,
        headers: &Headers,
        tables: &FTables,
        rconfig: &Config,
    ) -> CliResult<()> {
        let path = rconfig.path.as_ref().ok_or_else(|| {
            crate::CliError::Other("--frequency-jsonl requires a file input, not stdin".to_string())
        })?;

        let unique_headers = UNIQUE_COLUMNS_VEC.get().unwrap_or(&EMPTY_USIZE_VEC);
        let row_count = *FREQ_ROW_COUNT.get().unwrap_or(&0);
        if row_count == 0 {
            wwarn!(
                "--frequency-jsonl: row count is 0, skipping frequency cache (no data to cache). \
                 Run 'qsv stats --cardinality --stats-jsonl' first."
            );
            return Ok(());
        }
        let col_cardinality = COL_CARDINALITY_VEC.get().unwrap_or(&EMPTY_VEC);

        // Compute HIGH_CARDINALITY effective threshold
        #[allow(clippy::cast_precision_loss)]
        let high_card_pct_limit =
            (row_count as f64 * self.flag_high_card_pct as f64 / 100.0) as u64;
        let effective_threshold = (self.flag_high_card_threshold as u64).min(high_card_pct_limit);

        // Collect all column entries
        let mut entries: Vec<FrequencyCacheEntry> = Vec::with_capacity(headers.len());
        for (i, header) in headers.iter().enumerate() {
            let field_name = if rconfig.no_headers {
                (i + 1).to_string()
            } else {
                util::bytes_to_cow_str(header).into_owned()
            };

            // col_cardinality is aligned positionally to the selected columns, so index
            // by position (i) — not by name — so duplicate-named columns each get their
            // own cardinality.
            let cardinality = col_cardinality.get(i).copied().unwrap_or(0);

            let is_high_cardinality =
                !unique_headers.contains(&i) && cardinality > effective_threshold;

            let entry = if unique_headers.contains(&i) {
                // ALL_UNIQUE: single sentinel entry
                FrequencyCacheEntry {
                    field: field_name,
                    cardinality,
                    frequencies: vec![FrequencyCacheValue {
                        value:      "<ALL_UNIQUE>".to_string(),
                        count:      row_count,
                        percentage: 100.0,
                    }],
                }
            } else if is_high_cardinality {
                // HIGH_CARDINALITY: single sentinel entry
                FrequencyCacheEntry {
                    field: field_name,
                    cardinality,
                    frequencies: vec![FrequencyCacheValue {
                        value:      "<HIGH_CARDINALITY>".to_string(),
                        count:      row_count,
                        percentage: 100.0,
                    }],
                }
            } else if i < tables.len() {
                // Normal column: complete frequency data from FTable
                let ftab = &tables[i];
                let (counts_ref, total_count) = ftab.par_frequent(false); // descending
                let pct_factor = if total_count > 0 {
                    100.0 / total_count as f64
                } else {
                    0.0
                };

                let frequencies: Vec<FrequencyCacheValue> = counts_ref
                    .into_iter()
                    .map(|(value, count)| {
                        // Use empty string for null values in the cache so we can
                        // round-trip: FTable vec![] → cache "" → read vec![]
                        let val_str = if value.is_empty() {
                            String::new()
                        } else {
                            util::bytes_to_cow_str(value).into_owned()
                        };
                        FrequencyCacheValue {
                            value: val_str,
                            count,
                            percentage: count as f64 * pct_factor,
                        }
                    })
                    .collect();

                FrequencyCacheEntry {
                    field: field_name,
                    cardinality,
                    frequencies,
                }
            } else {
                continue;
            };

            entries.push(entry);
        }

        // Build metadata struct (written as first JSONL line)
        let metadata = FrequencyCacheMetadata {
            arg_input:                self.arg_input.clone().unwrap_or_default(),
            flag_high_card_threshold: self.flag_high_card_threshold,
            flag_high_card_pct:       self.flag_high_card_pct,
            flag_no_nulls:            self.flag_no_nulls,
            flag_no_headers:          self.flag_no_headers,
            flag_delimiter:           self
                .flag_delimiter
                .as_ref()
                .map_or_else(|| ",".to_string(), |d| (d.as_byte() as char).to_string()),
            record_count:             row_count,
            column_count:             headers.len(),
            date_generated:           chrono::Utc::now().to_rfc3339(),
            qsv_version:              env!("CARGO_PKG_VERSION").to_string(),
            selection_signature:      Self::selection_signature(headers),
            canonical_input_path:     self
                .arg_input
                .as_ref()
                .and_then(|p| fs::canonicalize(p).ok())
                .map(|p| p.to_string_lossy().into_owned())
                .unwrap_or_default(),
        };
        let num_cache_columns = entries.len();

        // Serialize as JSONL: metadata line + one entry per line
        let mut jsonl = String::with_capacity(entries.len() * 100 + 500); // rough estimate
        jsonl.push_str(&serde_json::to_string(&metadata)?);
        jsonl.push('\n');
        for entry in &entries {
            jsonl.push_str(&serde_json::to_string(entry)?);
            jsonl.push('\n');
        }

        let cache_path = Self::cache_path_for(path);
        let cache_len = jsonl.len();
        fs::write(&cache_path, jsonl)?;

        winfo!(
            "Frequency cache written: {} ({} bytes, {} columns, {} rows).",
            cache_path.display(),
            cache_len,
            num_cache_columns,
            row_count
        );

        Ok(())
    }

    /// Read and validate the frequency JSONL cache file.
    /// Returns None if cache doesn't exist, is stale, or is incompatible.
    /// On success, returns both the metadata (for selection-signature
    /// validation against the current invocation's selected headers) and the
    /// per-column entries.
    fn read_frequency_cache(
        &self,
        rconfig: &Config,
    ) -> Option<(FrequencyCacheMetadata, Vec<FrequencyCacheEntry>)> {
        use filetime::FileTime;

        let path = rconfig.path.as_ref()?;
        let cache_path = Self::cache_path_for(path);

        if !cache_path.exists() {
            log::info!("Frequency cache not found: {}", cache_path.display());
            return None;
        }

        // Compare mtime: cache must be newer than source CSV
        let csv_metadata = fs::metadata(path).ok()?;
        let cache_metadata = fs::metadata(&cache_path).ok()?;
        let csv_mtime = FileTime::from_last_modification_time(&csv_metadata);
        let cache_mtime = FileTime::from_last_modification_time(&cache_metadata);

        if cache_mtime <= csv_mtime {
            winfo!(
                "Frequency cache is stale (older than source CSV). Recomputing. Use \
                 --frequency-jsonl to regenerate."
            );
            return None;
        }

        // Read JSONL cache: line 1 = metadata, remaining lines = entries
        let jsonl_content = fs::read_to_string(&cache_path).ok()?;
        let mut lines = jsonl_content.lines();

        let metadata_line = lines.next()?;
        let metadata: FrequencyCacheMetadata = serde_json::from_str(metadata_line)
            .map_err(|e| {
                wwarn!("Failed to deserialize frequency cache metadata: {e}");
                e
            })
            .ok()?;

        let mut entries = Vec::new();
        for line in lines {
            if line.is_empty() {
                continue;
            }
            let entry: FrequencyCacheEntry = serde_json::from_str(line)
                .map_err(|e| {
                    wwarn!("Failed to deserialize frequency cache entry: {e}");
                    e
                })
                .ok()?;
            entries.push(entry);
        }

        // Validate cache args compatibility
        // flag_no_nulls affects what's stored in the FTable — mismatch means
        // cache has wrong null counts
        if metadata.flag_no_nulls != self.flag_no_nulls {
            winfo!(
                "Frequency cache incompatible: --no-nulls differs (cache={}, current={}). \
                 Recomputing.",
                metadata.flag_no_nulls,
                self.flag_no_nulls
            );
            return None;
        }
        // flag_no_headers affects column naming in the cache
        if metadata.flag_no_headers != self.flag_no_headers {
            winfo!(
                "Frequency cache incompatible: --no-headers differs (cache={}, current={}). \
                 Recomputing.",
                metadata.flag_no_headers,
                self.flag_no_headers
            );
            return None;
        }
        // flag_delimiter affects how values are parsed — mismatch means
        // cache has values split on different boundaries
        let current_delimiter = self
            .flag_delimiter
            .as_ref()
            .map_or_else(|| ",".to_string(), |d| (d.as_byte() as char).to_string());
        if metadata.flag_delimiter != current_delimiter {
            winfo!(
                "Frequency cache incompatible: --delimiter differs (cache={:?}, current={:?}). \
                 Recomputing.",
                metadata.flag_delimiter,
                current_delimiter
            );
            return None;
        }
        // Threshold differences don't invalidate the cache — the partial cache
        // mechanism handles the mismatch gracefully:
        //   - If the current threshold is MORE lenient (higher) than the cache, columns cached as
        //     HIGH_CARDINALITY will be recomputed via parallel computation.
        //   - If the current threshold is STRICTER (lower) than the cache, columns cached with full
        //     data will still be served from cache even though they would now be classified as
        //     HIGH_CARDINALITY. This is intentional: the cached data is correct and complete, just
        //     more detailed than the current threshold would produce on a fresh run. The threshold
        //     only controls which columns get full data vs a sentinel during cache *generation*,
        //     not reads.
        if metadata.flag_high_card_threshold != self.flag_high_card_threshold
            || metadata.flag_high_card_pct != self.flag_high_card_pct
        {
            log::info!(
                "Frequency cache threshold differs: cache=(threshold={}, pct={}), \
                 current=(threshold={}, pct={}). Partial cache still valid.",
                metadata.flag_high_card_threshold,
                metadata.flag_high_card_pct,
                self.flag_high_card_threshold,
                self.flag_high_card_pct,
            );
        }

        if entries.is_empty() {
            return None;
        }

        Some((metadata, entries))
    }

    /// Try to produce output directly from the frequency cache.
    /// Returns Ok(true) if output was produced from cache, Ok(false) if cache
    /// was not usable and normal computation should proceed.
    ///
    /// When all selected columns have cached data, `FTables` are reconstructed via
    /// `increment_by` and output is produced directly (full cache hit).
    ///
    /// When some columns have `HIGH_CARDINALITY` sentinels (no cached data), sets up a
    /// partial cache: pre-builds `FTables` for cached columns into `FREQ_CACHE_FTABLES`
    /// and marks them in `FREQ_CACHE_SKIP`. Returns Ok(false) so the normal parallel
    /// computation runs, but `ftables_unweighted` skips the cached columns (same
    /// pattern as `ALL_UNIQUE` skip). After computation, `run()` merges the cached
    /// `FTables` back into the result.
    #[allow(clippy::cast_precision_loss)]
    fn try_output_from_cache(&self, rconfig: &Config, is_json: bool) -> CliResult<bool> {
        // Read and validate the cache
        let Some((cache_metadata, cache_entries)) = self.read_frequency_cache(rconfig) else {
            return Ok(false);
        };

        // Read CSV headers to resolve --select
        let mut rdr = rconfig.reader()?;
        let full_headers = rdr.byte_headers()?.clone();

        // Safety net: can_use_freq_cache already guards flag_weight.is_none(),
        // but keep a runtime check because the selection below does not
        // exclude the weight column. If the guard in can_use_freq_cache is
        // ever loosened, this prevents silently including the weight column
        // in frequency output.
        if self.flag_weight.is_some() {
            return Ok(false);
        }
        let sel = self.rconfig().selection(&full_headers)?;
        let selected_headers: csv::ByteRecord = sel.select(&full_headers).collect();

        // Selection-signature validation. With --no-headers the cache keys
        // are positional (1..k within the current selection), so a cache
        // built with `--select 1,2` would silently match a `--select 2,1`
        // run and return the wrong columns. Compare the signature of the
        // current selected headers against the cache's recorded signature.
        // An empty recorded signature means the cache predates this field —
        // refuse it once and let it be regenerated.
        let current_sig = Self::selection_signature(&selected_headers);
        if cache_metadata.selection_signature.is_empty() {
            log::info!(
                "Frequency cache predates selection-signature validation. Recomputing (regenerate \
                 with --frequency-jsonl to refresh)."
            );
            return Ok(false);
        }
        if cache_metadata.selection_signature != current_sig {
            log::info!("Frequency cache incompatible: --select differs from cache. Recomputing.");
            return Ok(false);
        }

        let selected_col_names: Vec<String> = selected_headers
            .iter()
            .map(|h| {
                if rconfig.no_headers {
                    String::new() // handled by index below
                } else {
                    util::bytes_to_cow_str(h).into_owned()
                }
            })
            .collect();

        // Build a map from field name to cache entry for quick lookup
        let cache_map: foldhash::HashMap<&str, &FrequencyCacheEntry> = cache_entries
            .iter()
            .map(|e| (e.field.as_str(), e))
            .collect();
        // CSVs may have duplicate header names. The cache_map keeps only the last
        // entry per name, so duplicates would silently shadow. Warn — the behavior
        // is still well-defined (last-wins) but the user should know.
        if cache_map.len() != cache_entries.len() {
            log::warn!(
                "Frequency cache contains duplicate column names ({} entries, {} unique). Falling \
                 back to recomputation to avoid ambiguous lookups.",
                cache_entries.len(),
                cache_map.len()
            );
            return Ok(false);
        }

        // Filter cache entries to selected columns, preserving selection order
        let mut selected_entries: Vec<&FrequencyCacheEntry> =
            Vec::with_capacity(selected_col_names.len());

        if rconfig.no_headers {
            // With no-headers, columns are 1-based indices
            for (i, _) in selected_headers.iter().enumerate() {
                let col_name = (i + 1).to_string();
                if let Some(entry) = cache_map.get(col_name.as_str()) {
                    selected_entries.push(entry);
                } else {
                    log::info!("Column '{col_name}' not found in frequency cache, falling back");
                    return Ok(false);
                }
            }
        } else {
            for col_name in &selected_col_names {
                if let Some(entry) = cache_map.get(col_name.as_str()) {
                    selected_entries.push(entry);
                } else {
                    log::info!("Column '{col_name}' not found in frequency cache, falling back");
                    return Ok(false);
                }
            }
        }

        // Check if any selected column has HIGH_CARDINALITY sentinel.
        // If so, set up a partial cache: pre-build FTables for columns with actual
        // cached data, and let the parallel computation handle only the
        // HIGH_CARDINALITY columns (which have no data in the cache).
        let has_high_cardinality = selected_entries.iter().any(|entry| {
            entry.frequencies.len() == 1 && entry.frequencies[0].value == "<HIGH_CARDINALITY>"
        });

        if has_high_cardinality {
            let mut skip_vec: Vec<bool> = Vec::with_capacity(selected_entries.len());
            let mut cached_ftables: FTables = Vec::with_capacity(selected_entries.len());

            for entry in &selected_entries {
                let is_sentinel = entry.frequencies.len() == 1
                    && (entry.frequencies[0].value == "<HIGH_CARDINALITY>"
                        || entry.frequencies[0].value == "<ALL_UNIQUE>");

                if is_sentinel {
                    // No actual data in cache — needs computation (or ALL_UNIQUE skip)
                    skip_vec.push(false);
                    cached_ftables.push(Frequencies::with_capacity(1));
                } else {
                    // Reconstruct FTable from cached data
                    skip_vec.push(true);
                    let mut ftab: FTable = Frequencies::with_capacity(entry.frequencies.len());
                    for freq_val in &entry.frequencies {
                        let key: Vec<u8> = if freq_val.value.is_empty() {
                            Vec::new()
                        } else {
                            freq_val.value.as_bytes().to_vec()
                        };
                        ftab.increment_by(key, freq_val.count);
                    }
                    cached_ftables.push(ftab);
                }
            }

            let cached_count = skip_vec.iter().filter(|&&s| s).count();
            winfo!(
                "Partial frequency cache hit: {cached_count}/{} columns from cache, {} need \
                 computation.",
                skip_vec.len(),
                skip_vec.len() - cached_count,
            );

            if FREQ_CACHE_SKIP.set(skip_vec).is_err() {
                log::warn!("FREQ_CACHE_SKIP already set — stale partial cache may be used");
            }
            if FREQ_CACHE_FTABLES.set(cached_ftables).is_err() {
                log::warn!("FREQ_CACHE_FTABLES already set — stale partial cache may be used");
            }
            return Ok(false);
        }

        // Derive row_count from cache: sum counts of first non-ALL_UNIQUE column,
        // or use cardinality of an ALL_UNIQUE column
        let row_count = selected_entries
            .iter()
            .find_map(|entry| {
                if entry.frequencies.len() == 1 && entry.frequencies[0].value == "<ALL_UNIQUE>" {
                    Some(entry.cardinality)
                } else {
                    let total: u64 = entry.frequencies.iter().map(|f| f.count).sum();
                    if total > 0 { Some(total) } else { None }
                }
            })
            .unwrap_or(0);

        if row_count == 0 {
            return Ok(false);
        }

        // Initialize OnceLocks that the output pipeline needs
        NULL_VAL
            .get()
            .expect("NULL_VAL should already be set in run()");

        if FREQ_ROW_COUNT.set(row_count).is_err() {
            log::warn!("FREQ_ROW_COUNT already set — stale row count may be used");
        }

        // Build unique columns vec: columns where cache has ALL_UNIQUE sentinel
        let unique_columns: Vec<usize> = selected_entries
            .iter()
            .enumerate()
            .filter(|(_, entry)| {
                entry.frequencies.len() == 1 && entry.frequencies[0].value == "<ALL_UNIQUE>"
            })
            .map(|(i, _)| i)
            .collect();
        if UNIQUE_COLUMNS_VEC.set(unique_columns).is_err() {
            log::warn!("UNIQUE_COLUMNS_VEC already set — stale unique columns may be used");
        }

        // Reconstruct FTables from cache using increment_by
        let mut tables: FTables = Vec::with_capacity(selected_entries.len());
        for entry in &selected_entries {
            if entry.frequencies.len() == 1 && entry.frequencies[0].value == "<ALL_UNIQUE>" {
                // ALL_UNIQUE: push empty FTable, process_frequencies handles the sentinel
                tables.push(Frequencies::with_capacity(1));
                continue;
            }
            let mut ftab: FTable = Frequencies::with_capacity(entry.frequencies.len());
            for freq_val in &entry.frequencies {
                // Convert empty string back to empty vec (null representation)
                let key: Vec<u8> = if freq_val.value.is_empty() {
                    Vec::new()
                } else {
                    freq_val.value.as_bytes().to_vec()
                };
                ftab.increment_by(key, freq_val.count);
            }
            tables.push(ftab);
        }

        // Now produce output using the existing pipeline
        // can_use_freq_cache already guards !is_json, so this is unreachable
        debug_assert!(!is_json, "try_output_from_cache called with JSON mode");

        // CSV output mode — reuse the shared unweighted emit helper.
        let abs_dec_places = self.flag_pct_dec_places.unsigned_abs() as u32;
        // safety: we just set UNIQUE_COLUMNS_VEC above
        let unique_headers_vec = UNIQUE_COLUMNS_VEC.get().unwrap();

        let mut wtr = Config::new(self.flag_output.as_ref()).writer()?;
        wtr.write_record(vec!["field", "value", "count", "percentage", "rank"])?;

        self.emit_unweighted_csv_rows(
            &mut wtr,
            &selected_headers,
            tables,
            unique_headers_vec,
            abs_dec_places,
            row_count,
            rconfig.no_headers,
        )?;
        wtr.flush()?;

        winfo!("Frequency cache hit: output produced from cache.");
        Ok(true)
    }

    /// Helper to move "Other" category to end if not sorted
    fn move_other_to_end_if_needed<T>(&self, counts: &mut [(Vec<u8>, T, f64, f64)]) {
        let other_prefix = format!("{} (", self.flag_other_text);
        let other_prefix_bytes = other_prefix.as_bytes();
        if !self.flag_other_sorted
            && counts
                .first()
                .is_some_and(|(value, _, _, _)| value.starts_with(other_prefix_bytes))
        {
            counts.rotate_left(1);
        }
    }

    /// Helper to move NULL category to end if not sorted.
    /// Unlike `move_other_to_end_if_needed()` which only checks position 0 (since "Other" always
    /// has rank 0 and appears first in ascending sort), NULL can appear anywhere based on its
    /// count, so we need to search the entire list.
    /// This function handles multiple NULL entries (e.g., when literal "(NULL)" values exist
    /// in the data alongside empty strings that are converted to "(NULL)").
    fn move_null_to_end_if_needed<T: Copy>(&self, counts: &mut Vec<(Vec<u8>, T, f64, f64)>) {
        if self.flag_null_sorted {
            return;
        }
        // safety: NULL_VAL is set in run()
        let null_val = NULL_VAL.get().unwrap();

        // Single-pass stable extraction: keeps non-NULL entries in their existing
        // (already sorted) order, collects NULL entries, and appends them at the end.
        // Replaces an earlier O(n²) Vec::remove loop that was hot for high-cardinality
        // columns with multiple NULL entries.
        let null_entries: Vec<_> = counts
            .extract_if(.., |entry| entry.0.as_slice() == null_val.as_slice())
            .collect();
        counts.extend(null_entries);
    }

    /// Process weighted frequencies
    fn process_frequencies_weighted(
        &self,
        all_unique_header: bool, /* Indicates the column has all-unique values (e.g., an ID
                                  * column). For all-unique columns, show a single <ALL_UNIQUE>
                                  * entry with the sum of all weights. */
        abs_dec_places: u32,
        _row_count: u64,
        weighted_map: &HashMap<Vec<u8>, f64>,
        processed_frequencies: &mut Vec<ProcessedFrequency>,
    ) {
        if all_unique_header {
            // For all-unique headers with weighted frequencies, create a single entry
            // with the sum of all weights
            let total_weight: f64 = weighted_map.values().sum();
            // Skip emitting an all-unique entry if the total weight is non-finite (NaN or infinity)
            // to avoid producing a misleading count, consistent with the non-all-unique path
            // where non-finite weights are skipped entirely.
            if !total_weight.is_finite() {
                return;
            }
            // u64::MAX as f64 rounds up past u64::MAX, but Rust's saturating
            // f64-to-int cast (since 1.45) clamps down on conversion, so the
            // result is never UB. The explicit clamp also pins NaN to 0.0
            // (which the earlier is_finite check already rejects).
            #[allow(clippy::cast_precision_loss)]
            let count = total_weight.clamp(0.0, u64::MAX as f64).round() as u64;
            processed_frequencies.push(ProcessedFrequency {
                value: ALL_UNIQUE_TEXT.get().unwrap().clone(),
                count,
                percentage: 100.0,
                formatted_percentage: self.format_percentage(100.0, abs_dec_places),
                rank: 0.0, // Rank 0 for all-unique headers
            });
            return;
        }

        // For non-all-unique columns, process individual weighted values
        let mut counts_to_process = self.counts_weighted(weighted_map);
        self.move_other_to_end_if_needed(&mut counts_to_process);
        self.move_null_to_end_if_needed(&mut counts_to_process);

        // Convert to processed frequencies (count is f64, convert to u64 for display)
        for (value, weight, percentage, rank) in counts_to_process {
            // Skip non-finite weights (NaN or infinity) to avoid emitting misleading zero-count
            // rows.
            if !weight.is_finite() {
                continue;
            }
            #[allow(clippy::cast_precision_loss)]
            let count = weight.clamp(0.0, u64::MAX as f64).round() as u64;
            processed_frequencies.push(ProcessedFrequency {
                value,
                count,
                percentage,
                formatted_percentage: self.format_percentage(percentage, abs_dec_places),
                rank,
            });
        }
    }

    /// Shared frequency processing function used by both CSV and JSON output
    fn process_frequencies(
        &self,
        all_unique_header: bool,
        abs_dec_places: u32,
        row_count: u64,
        ftab: &FTable,
        processed_frequencies: &mut Vec<ProcessedFrequency>,
    ) {
        if all_unique_header {
            // For all-unique headers, create a single entry
            processed_frequencies.push(ProcessedFrequency {
                value:                ALL_UNIQUE_TEXT.get().unwrap().clone(),
                count:                row_count,
                percentage:           100.0,
                formatted_percentage: self.format_percentage(100.0, abs_dec_places),
                rank:                 0.0, // Rank 0 for all-unique headers
            });
        } else {
            // Process regular frequencies
            let mut counts_to_process = self.counts(ftab);
            self.move_other_to_end_if_needed(&mut counts_to_process);
            self.move_null_to_end_if_needed(&mut counts_to_process);

            // Convert to processed frequencies
            for (value, count, percentage, rank) in counts_to_process {
                processed_frequencies.push(ProcessedFrequency {
                    value,
                    count,
                    percentage,
                    formatted_percentage: self.format_percentage(percentage, abs_dec_places),
                    rank,
                });
            }
        }
    }

    /// Emit the unweighted CSV body for one column-set: writes one row per
    /// processed frequency for every (header, ftable) pair.
    ///
    /// Shared between `run()` (post-computation) and `try_output_from_cache()`
    /// (full cache hit). Caller is responsible for writing the column-header
    /// record before this and flushing afterward.
    fn emit_unweighted_csv_rows<W: io::Write>(
        &self,
        wtr: &mut csv::Writer<W>,
        headers: &Headers,
        tables: FTables,
        unique_headers_vec: &[usize],
        abs_dec_places: u32,
        row_count: u64,
        no_headers: bool,
    ) -> CliResult<()> {
        let vis_whitespace = self.flag_vis_whitespace;
        let mut header_vec: Vec<u8>;
        let mut itoa_buffer = itoa::Buffer::new();
        let mut zmij_buffer = zmij::Buffer::new();
        let mut rank_buffer = String::with_capacity(20);
        #[allow(unused_assignments)]
        let mut value_str = String::with_capacity(100);
        let mut processed_frequencies: Vec<ProcessedFrequency> = Vec::with_capacity(headers.len());

        for (i, (header, ftab)) in headers.iter().zip(tables).enumerate() {
            header_vec = if no_headers {
                (i + 1).to_string().into_bytes()
            } else {
                header.to_vec()
            };

            self.process_frequencies(
                unique_headers_vec.contains(&i),
                abs_dec_places,
                row_count,
                &ftab,
                &mut processed_frequencies,
            );

            for processed_freq in &processed_frequencies {
                // Format rank: show as integer if whole number, otherwise with decimals.
                // Sentinel value -1.0 indicates NULL entry with --pct-nulls=false (empty rank).
                rank_buffer.clear();
                if processed_freq.rank >= 0.0 {
                    if processed_freq.rank.fract() == 0.0 {
                        rank_buffer.push_str(itoa_buffer.format(processed_freq.rank as u64));
                    } else {
                        rank_buffer.push_str(zmij_buffer.format(processed_freq.rank));
                    }
                }

                let value_bytes: &[u8] = if vis_whitespace {
                    value_str =
                        util::visualize_whitespace(&util::bytes_to_cow_str(&processed_freq.value));
                    value_str.as_bytes()
                } else {
                    &processed_freq.value
                };

                wtr.write_record([
                    &*header_vec,
                    value_bytes,
                    itoa_buffer.format(processed_freq.count).as_bytes(),
                    processed_freq.formatted_percentage.as_bytes(),
                    rank_buffer.as_bytes(),
                ])?;
            }
            processed_frequencies.clear();
        }
        Ok(())
    }

    /// Format percentage with proper decimal places
    fn format_percentage(&self, percentage: f64, abs_dec_places: u32) -> String {
        // Sentinel value -1.0 indicates NULL entry with --pct-nulls=false
        if percentage < 0.0 {
            return String::new();
        }
        let pct_decimal = Decimal::from_f64(percentage).unwrap_or_default();
        let pct_scale = if self.flag_pct_dec_places < 0 {
            let current_scale = pct_decimal.scale();
            current_scale.max(abs_dec_places)
        } else {
            abs_dec_places
        };
        let final_pct_decimal = pct_decimal
            .round_dp_with_strategy(
                pct_scale,
                rust_decimal::RoundingStrategy::MidpointAwayFromZero,
            )
            .normalize();
        // Optimize: Check scale directly instead of converting to string for length check
        if final_pct_decimal.scale() > abs_dec_places {
            final_pct_decimal
                .round_dp_with_strategy(abs_dec_places, RoundingStrategy::MidpointAwayFromZero)
                .normalize()
                .to_string()
        } else {
            final_pct_decimal.to_string()
        }
    }

    /// Process weighted frequencies from `HashMap` and return same format as `counts()`
    #[allow(clippy::cast_precision_loss)]
    fn counts_weighted(
        &self,
        weighted_map: &HashMap<Vec<u8>, f64>,
    ) -> Vec<(ByteString, f64, f64, f64)> {
        // Convert HashMap to Vec and sort
        let mut counts: Vec<(Vec<u8>, f64)> =
            weighted_map.iter().map(|(k, v)| (k.clone(), *v)).collect();

        // Sort by count (weight), breaking ties by value for deterministic output
        if self.flag_asc {
            counts.sort_unstable_by(|a, b| {
                a.1.partial_cmp(&b.1)
                    .unwrap_or(std::cmp::Ordering::Equal)
                    .then_with(|| a.0.cmp(&b.0))
            });
        } else {
            counts.sort_unstable_by(|a, b| {
                b.1.partial_cmp(&a.1)
                    .unwrap_or(std::cmp::Ordering::Equal)
                    .then_with(|| a.0.cmp(&b.0))
            });
        }

        // Calculate total weight (sum of all weights)
        let total_weight: f64 = weighted_map.values().sum();

        // When --pct-nulls is false, extract NULL entries before ranking
        // so that non-NULL values get correct ranks (excluding NULLs from ranking)
        let null_entry = if self.flag_pct_nulls {
            None
        } else {
            counts
                .iter()
                .position(|(k, _)| k.is_empty())
                .map(|pos| counts.remove(pos))
        };

        // Calculate NULL weight for adjusted percentage
        let null_weight = null_entry.as_ref().map_or(0.0, |(_, w)| *w);

        // Apply limits
        let unique_counts_len = counts.len();
        apply_limits_weighted(&mut counts, self.flag_limit, self.flag_lmt_threshold);

        // Calculate pct_factor: when --pct-nulls is false, exclude NULLs from denominator
        let adjusted_total = total_weight - null_weight;
        let pct_factor = if adjusted_total > 0.0 {
            100.0_f64 / adjusted_total
        } else {
            0.0_f64
        };

        // Compute tolerance once before the loop (outside hot path)
        // Use stats cache if available to determine weight scale for more accurate tolerance
        let weight_tolerance = if let Some(ref weight_col) = self.flag_weight {
            STATS_RECORDS
                .get()
                .and_then(|records| records.get(weight_col))
                .and_then(|stats| {
                    // Prefer stddev as it represents the scale of variation
                    // Fall back to range or mean if stddev not available
                    stats
                        .stddev
                        .or(stats.range)
                        .or(stats.mean)
                        .filter(|&s| s > 0.0)
                })
                // Use a scale-aware tolerance with a minimum absolute epsilon to handle
                // both small and large weight scales robustly.
                .map_or(f64::EPSILON, |scale| (scale * 1e-6).max(1e-10))
        } else {
            f64::EPSILON
        };

        // Group by weight to handle ties
        let weight_groups = group_by_weight(counts, weight_tolerance);

        // safety: NULL_VAL is set in the main function
        let null_val = NULL_VAL.get().unwrap();

        // Apply ranking strategy (NULLs already removed when pct_nulls is false)
        let (mut counts_final, count_sum, pct_sum) = apply_ranking_strategy_weighted(
            weight_groups,
            self.flag_rank_strategy,
            pct_factor,
            null_val,
            self.flag_pct_nulls,
        );

        // Add NULL entry back with sentinel values when --pct-nulls is false
        // Insert at the correct sorted position based on weight to preserve sort order
        if let Some((_, null_weight_val)) = null_entry {
            let null_entry_final = (null_val.to_vec(), null_weight_val, -1.0, -1.0);
            // Find the correct insertion position using binary search (O(log n))
            // since counts_final is already sorted by weight.
            // safety: partition_point requires the slice to be partitioned by the predicate,
            // which holds because counts_final is sorted and all weights are finite
            // (NaN/Inf/non-positive weights are filtered out during accumulation).
            debug_assert!(
                null_weight_val.is_finite(),
                "null_weight_val must be finite for partition_point"
            );
            debug_assert!(
                counts_final.iter().all(|(_, w, _, _)| w.is_finite()),
                "all weights must be finite for partition_point"
            );
            debug_assert!(
                if self.flag_asc {
                    counts_final.array_windows::<2>().all(|[a, b]| a.1 <= b.1)
                } else {
                    counts_final.array_windows::<2>().all(|[a, b]| a.1 >= b.1)
                },
                "counts_final must be sorted by weight for partition_point"
            );
            let insert_pos = if self.flag_asc {
                // Ascending: find first position where weight >= null_weight
                // (places null before entries with equal weight)
                counts_final.partition_point(|(_, w, _, _)| *w < null_weight_val)
            } else {
                // Descending: find first position where weight < null_weight
                // (places null after entries with equal weight)
                counts_final.partition_point(|(_, w, _, _)| *w >= null_weight_val)
            };
            counts_final.insert(insert_pos, null_entry_final);
        }

        // Calculate "Other" category
        // When NULL was extracted, adjust calculations to exclude it
        let (adjusted_other_weight, adjusted_unique_len) = if null_weight > 0.0 {
            // Subtract null_weight from other_weight since NULL is handled separately
            (
                total_weight - count_sum - null_weight,
                unique_counts_len.saturating_sub(1), // NULL was removed from counts
            )
        } else {
            (total_weight - count_sum, unique_counts_len)
        };

        // When NULL was extracted and re-added, don't count it as a "shown" entry
        // because it's separate from the top-k values
        let shown_count = if null_weight > 0.0 {
            counts_final.len().saturating_sub(1)
        } else {
            counts_final.len()
        };
        let other_unique_count = adjusted_unique_len.saturating_sub(shown_count);
        // Only create Other entry if there are actually remaining unique values
        // and the weight is positive. This prevents "Other (0)" entries when
        // --limit 0 is used and all values are included.
        // Use 100.0 - pct_sum to ensure percentages sum exactly to 100%,
        // handling floating-point precision issues consistently with unweighted case.
        if adjusted_other_weight > 0.0 && other_unique_count > 0 && self.flag_other_text != "<NONE>"
        {
            counts_final.push((
                format!(
                    "{} ({})",
                    self.flag_other_text,
                    HumanCount(other_unique_count as u64)
                )
                .as_bytes()
                .to_vec(),
                adjusted_other_weight,
                100.0_f64 - pct_sum,
                0.0,
            ));
        }

        counts_final
    }

    #[inline]
    fn counts(&self, ftab: &FTable) -> Vec<(ByteString, u64, f64, f64)> {
        // Total count and cardinality straight from the table (cheap, no sort).
        // Both include the NULL/empty-key bucket, matching par_frequent's totals.
        let total_count = ftab.total_count();
        let unique_counts_len = ftab.len();

        // Fast path: for a positive --limit N below the cardinality, a heap-based
        // top_n/bottom_n selects the N kept values in O(k log N) instead of fully
        // sorting all k unique values (O(k log k)) then truncating. We only take it
        // when a plain top-N truncation is exactly what apply_limits_unweighted
        // would do, so we stay on par_frequent for: limit <= 0 and below-threshold
        // columns.
        let abs_limit = self.flag_limit.unsigned_abs();

        // A column is all-unique (every count == 1) iff its total count equals its
        // cardinality. Only then does --unq-limit change the effective truncation:
        // apply_limits_unweighted then keeps min(unq_limit, abs_limit) values.
        let all_unique = total_count == unique_counts_len as u64;
        let unq_active = all_unique && self.flag_unq_limit != abs_limit && self.flag_unq_limit > 0;
        let effective_limit = if unq_active {
            self.flag_unq_limit.min(abs_limit)
        } else {
            abs_limit
        };

        let threshold_fires =
            self.flag_lmt_threshold == 0 || self.flag_lmt_threshold >= unique_counts_len;
        let use_topn =
            self.flag_limit > 0 && threshold_fires && effective_limit < unique_counts_len;

        let (counts, null_entry): (Vec<(Vec<u8>, u64)>, Option<(Vec<u8>, u64)>) = if use_topn {
            // When --pct-nulls is false, NULL is reported separately and does not
            // consume a limit slot. Since at most one NULL bucket exists, fetching
            // effective_limit + 1 and dropping a selected NULL yields exactly the
            // non-NULL top-effective_limit (matching "sort all -> remove NULL ->
            // truncate").
            let want_null = !self.flag_pct_nulls;
            let fetch_n = effective_limit + usize::from(want_null);
            let selected = if self.flag_asc {
                ftab.bottom_n(fetch_n)
            } else {
                ftab.top_n(fetch_n)
            };
            let mut counts: Vec<(Vec<u8>, u64)> =
                selected.into_iter().map(|(k, v)| (k.clone(), v)).collect();

            let null_entry = if want_null {
                if let Some(pos) = counts.iter().position(|(k, _)| k.is_empty()) {
                    counts.remove(pos);
                }
                // Fetch the NULL count directly so a rare NULL outside the
                // fetched top-N is still reported.
                let nc = ftab.count(&Vec::new());
                (nc > 0).then(|| (Vec::new(), nc))
            } else {
                None
            };
            counts.truncate(effective_limit);
            (counts, null_entry)
        } else {
            // Full-sort path: preserves behavior for limit <= 0, negative limits,
            // and below-threshold columns.
            let counts_ref = if self.flag_asc {
                // ascending order - least frequent values first
                ftab.par_frequent(true).0
            } else {
                // descending order - most frequent values first
                ftab.par_frequent(false).0
            };
            let mut counts: Vec<(Vec<u8>, u64)> = counts_ref
                .into_iter()
                .map(|(k, v)| (k.clone(), v))
                .collect();

            // When --pct-nulls is false, extract NULL entries before ranking
            // so that non-NULL values get correct ranks (excluding NULLs from ranking)
            let null_entry = if self.flag_pct_nulls {
                None
            } else {
                // Find and remove NULL entry from counts
                counts
                    .iter()
                    .position(|(k, _)| k.is_empty())
                    .map(|pos| counts.remove(pos))
            };

            // Apply limits (including unique limit logic for all-unique columns)
            apply_limits_unweighted(
                &mut counts,
                self.flag_limit,
                self.flag_unq_limit,
                self.flag_lmt_threshold,
                all_unique,
            );
            (counts, null_entry)
        };

        // Calculate NULL count for adjusted percentage
        let null_count = null_entry.as_ref().map_or(0, |(_, c)| *c);

        // Calculate pct_factor: when --pct-nulls is false, exclude NULLs from denominator
        let adjusted_total = total_count.saturating_sub(null_count);
        let pct_factor = if adjusted_total > 0 {
            100.0_f64 / adjusted_total.to_f64().unwrap_or(1.0_f64)
        } else {
            0.0_f64
        };

        // Group by count to handle ties
        let count_groups = group_by_count(counts);

        // safety: NULL_VAL is set in the main function
        let null_val = NULL_VAL.get().unwrap();

        // Apply ranking strategy (NULLs already removed when pct_nulls is false)
        let (mut counts_final, count_sum, pct_sum) = apply_ranking_strategy_unweighted(
            count_groups,
            self.flag_rank_strategy,
            pct_factor,
            null_val,
            self.flag_pct_nulls,
        );

        // Add NULL entry back with sentinel values when --pct-nulls is false
        // Insert at the correct sorted position based on count to preserve sort order
        if let Some((_, null_count_val)) = null_entry {
            let null_entry_final = (null_val.to_vec(), null_count_val, -1.0, -1.0);
            // Find the correct insertion position using binary search (O(log n))
            // since counts_final is already sorted by count.
            // safety: partition_point requires the slice to be partitioned by the predicate,
            // which holds because counts_final is sorted and counts are u64 (always finite).
            debug_assert!(
                if self.flag_asc {
                    counts_final.array_windows::<2>().all(|[a, b]| a.1 <= b.1)
                } else {
                    counts_final.array_windows::<2>().all(|[a, b]| a.1 >= b.1)
                },
                "counts_final must be sorted by count for partition_point"
            );
            let insert_pos = if self.flag_asc {
                // Ascending: find first position where count >= null_count
                // (places null before entries with equal count)
                counts_final.partition_point(|(_, c, _, _)| *c < null_count_val)
            } else {
                // Descending: find first position where count < null_count
                // (places null after entries with equal count)
                counts_final.partition_point(|(_, c, _, _)| *c >= null_count_val)
            };
            counts_final.insert(insert_pos, null_entry_final);
        }

        // Calculate "Other" category
        // When NULL was extracted, adjust calculations to exclude it
        let (adjusted_other_count, adjusted_unique_len) = if null_count > 0 {
            // Subtract null_count from other_count since NULL is handled separately
            (
                total_count
                    .saturating_sub(count_sum)
                    .saturating_sub(null_count),
                unique_counts_len.saturating_sub(1), // NULL was removed from counts
            )
        } else {
            (total_count - count_sum, unique_counts_len)
        };

        // When NULL was extracted and re-added, don't count it as a "shown" entry
        // because it's separate from the top-k values
        let shown_count = if null_count > 0 {
            counts_final.len().saturating_sub(1)
        } else {
            counts_final.len()
        };
        let other_unique_count = adjusted_unique_len.saturating_sub(shown_count);
        // Only emit Other when there are remaining unique values AND remaining count.
        // The other_unique_count > 0 guard prevents misleading "Other (0)" rows
        // (e.g., with --limit 0 when all values are already shown), matching the
        // weighted path.
        if adjusted_other_count > 0 && other_unique_count > 0 && self.flag_other_text != "<NONE>" {
            counts_final.push((
                format!(
                    "{} ({})",
                    self.flag_other_text,
                    HumanCount(other_unique_count as u64)
                )
                .as_bytes()
                .to_vec(),
                adjusted_other_count,
                100.0_f64 - pct_sum,
                0.0, // Special rank for "Other" category
            ));
        }

        counts_final
    }

    pub fn sequential_ftables(&self) -> CliResult<(Headers, FTables, Option<WeightedFTables>)> {
        let mut rdr = self.rconfig().reader()?;
        let (headers, sel, weight_col_idx) = self.sel_headers(&mut rdr)?;
        if weight_col_idx.is_some() {
            let weighted =
                self.ftables_weighted_internal(&sel, rdr.byte_records(), 1, weight_col_idx);
            Ok((headers, vec![], Some(weighted)))
        } else {
            Ok((
                headers,
                self.ftables_unweighted(&sel, rdr.byte_records(), 1),
                None,
            ))
        }
    }

    // `rconfig` must be the SAME (already-resolved) Config that produced `idx`.
    // For special-format inputs (e.g. `.zip`) each `Config` lazily resolves to its
    // own temp path; cloning `rconfig` shares its `Arc<OnceLock>` so every worker
    // re-opens the index next to the same temp. Recreating `args.rconfig()` inside
    // a worker would resolve to a different temp with no sibling `.idx` and panic.
    pub fn parallel_ftables(
        &self,
        idx: &Indexed<fs::File, fs::File>,
        rconfig: &Config,
    ) -> CliResult<(Headers, FTables, Option<WeightedFTables>)> {
        let mut rdr = rconfig.reader()?;
        let (headers, sel, weight_col_idx) = self.sel_headers(&mut rdr)?;

        let idx_count = idx.count() as usize;
        if idx_count == 0 {
            return Ok((headers, vec![], None));
        }

        let njobs = util::njobs(self.flag_jobs);

        // Retain freed jemalloc pages for the duration of this parallel,
        // hashmap-heavy ftables build to avoid madvise page-return churn (no-op
        // when background_thread is active or QSV_NO_ALLOC_TUNING is set). Gated on
        // the genuine parallel path — sequential frequency counting does not
        // benefit and would only pay the extra RSS. Reached by both `frequency`
        // and `schema` (which calls this directly).
        if njobs > 1 {
            util::retain_alloc_pages_for_aggregation();
        }

        // Read memory limit from environment variable.
        // If QSV_FREQ_CHUNK_MEMORY_MB is set & valid, set max chunk memory
        // If QSV_FREQ_CHUNK_MEMORY_MB is not set, use 0 (dynamic sizing)
        // If QSV_FREQ_CHUNK_MEMORY_MB is set to a value that cannot be parsed as u64 (e.g., -1 or
        // any invalid/non-positive value), use CPU-based chunking
        let max_chunk_memory_mb = if let Ok(val) = std::env::var("QSV_FREQ_CHUNK_MEMORY_MB") {
            // if valid, set max chunk memory
            // if invalid (cannot be parsed as u64), use CPU-based chunking
            atoi_simd::parse::<u64, false, false>(val.as_bytes()).ok()
        } else {
            Some(0) // default to dynamic sizing
        };

        // Sample first 1000 records for memory estimation
        // Always sample when QSV_FREQ_CHUNK_MEMORY_MB is set to ANY value (including 0)
        // or when not set (to enable dynamic sizing)
        let sample_records = if max_chunk_memory_mb.is_some() {
            util::sample_records(rconfig, 1000)
        } else {
            None
        };

        let (chunking_mode, chunk_size) = if let Some(limit_mb) = max_chunk_memory_mb {
            // Calculate memory-aware chunk size
            let chunk_size = calculate_memory_aware_chunk_size_for_frequency(
                idx_count as u64,
                njobs,
                max_chunk_memory_mb,
                sample_records.as_deref(),
            );

            // Log chunk size and memory estimates for debugging
            // Log when memory-aware chunking is active (either explicitly set or automatically
            // enabled) Estimate average record size from samples if available
            let avg_record_size = if let Some(samples) = sample_records {
                calculate_avg_record_size_for_frequency(&samples)
            } else {
                1024 // Default: 1KB per record
            };

            let estimated_memory_mb =
                estimate_chunk_memory_for_frequency(chunk_size, avg_record_size, headers.len())
                    / (1024 * 1024);
            let chunking_mode = if limit_mb == 0 {
                "dynamic (auto)"
            } else {
                "fixed limit"
            };
            (
                format!(
                    "Memory-aware chunking ({chunking_mode}): chunk_size={chunk_size}, \
                     estimated_memory_mb={estimated_memory_mb:.2}"
                ),
                chunk_size,
            )
        } else {
            let chunk_size = util::chunk_size(idx_count, njobs);
            (
                format!("CPU-based chunking: chunk_size={chunk_size}"),
                chunk_size,
            )
        };

        let nchunks = util::num_of_chunks(idx_count, chunk_size);
        log::info!("({chunking_mode}) nchunks={nchunks}");

        if weight_col_idx.is_some() {
            // Parallel weighted frequencies
            let pool = ThreadPool::new(njobs);
            let (send, recv) = crossbeam_channel::bounded(nchunks);
            for i in 0..nchunks {
                let (send, args, rconf, sel, weight_idx) = (
                    send.clone(),
                    self.clone(),
                    rconfig.clone(),
                    sel.clone(),
                    weight_col_idx,
                );
                pool.execute(move || {
                    let mut idx = rconf.indexed().unwrap().unwrap();
                    idx.seek((i * chunk_size) as u64).unwrap();
                    let it = idx.byte_records().take(chunk_size);
                    send.send(args.ftables_weighted_internal(&sel, it, nchunks, weight_idx))
                        .unwrap();
                });
            }
            drop(send);
            // Parallel reduce of partial WeightedFTables. Use `par_bridge` so
            // the reducer can consume chunks as they arrive instead of
            // materializing all of them — peak memory is bounded by the
            // rayon pool's working set, not by `nchunks` (which can grow
            // large when memory-aware chunking picks small chunks).
            let merged = recv
                .into_iter()
                .par_bridge()
                .reduce(Vec::new, merge_weighted_ftables);
            Ok((headers, vec![], Some(merged)))
        } else {
            // Parallel unweighted frequencies
            let pool = ThreadPool::new(njobs);
            let (send, recv) = crossbeam_channel::bounded(nchunks);
            for i in 0..nchunks {
                let (send, args, rconf, sel) =
                    (send.clone(), self.clone(), rconfig.clone(), sel.clone());
                pool.execute(move || {
                    let mut idx = rconf.indexed().unwrap().unwrap();
                    idx.seek((i * chunk_size) as u64).unwrap();
                    let it = idx.byte_records().take(chunk_size);
                    send.send(args.ftables_unweighted(&sel, it, nchunks))
                        .unwrap();
                });
            }
            drop(send);
            // Parallel reduce of partial FTables. Use `par_bridge` so the
            // reducer can consume chunks as they arrive instead of
            // materializing all of them — peak memory is bounded by the
            // rayon pool's working set, not by `nchunks`.
            let merged = recv
                .into_iter()
                .par_bridge()
                .reduce(Vec::new, merge_ftables);
            Ok((headers, merged, None))
        }
    }

    #[inline]
    fn ftables_weighted_internal<I>(
        &self,
        sel: &Selection,
        it: I,
        nchunks: usize,
        weight_col_idx: Option<usize>,
    ) -> WeightedFTables
    where
        I: Iterator<Item = csv::Result<csv::ByteRecord>>,
    {
        let sel_len = sel.len();

        let mut field_buffer: Vec<u8> = Vec::with_capacity(1024);
        let mut row_buffer: csv::ByteRecord = csv::ByteRecord::with_capacity(200, sel_len);
        let mut string_buf = String::with_capacity(512);

        // For weighted frequencies, we process all columns including all-unique ones
        // because the weights provide meaningful information, so we don't need to track
        // which columns are all-unique here (unlike unweighted frequencies where all-unique
        // columns are skipped for memory efficiency)
        let flag_no_nulls = self.flag_no_nulls;
        let flag_ignore_case = self.flag_ignore_case;
        let flag_no_trim = self.flag_no_trim;

        let col_cardinality_vec = COL_CARDINALITY_VEC.get().unwrap_or(&EMPTY_VEC);
        let mut weighted_freq_tables: Vec<HashMap<Vec<u8>, f64>> = if col_cardinality_vec.is_empty()
        {
            (0..sel_len).map(|_| HashMap::with_capacity(1000)).collect()
        } else {
            (0..sel_len)
                .map(|i| {
                    // For weighted frequencies, we process all columns including all-unique ones
                    // so we use the actual cardinality (or a reasonable default) rather than 1
                    let capacity = if nchunks == 1 {
                        col_cardinality_vec
                            .get(i)
                            .map_or(1000, |&cardinality| cardinality as usize)
                    } else {
                        let cardinality = col_cardinality_vec
                            .get(i)
                            .map_or(1000, |&cardinality| cardinality as usize);
                        cardinality / nchunks
                    };
                    HashMap::with_capacity(capacity)
                })
                .collect()
        };

        // Helper to do borrowed-key insert/increment on HashMap<Vec<u8>, f64>.
        // Uses get_mut(&[u8]) first to avoid allocating when key already exists.
        #[allow(clippy::items_after_statements)]
        #[inline]
        fn weighted_add(map: &mut HashMap<Vec<u8>, f64>, key: &[u8], weight: f64) {
            // Upstream validation (weight.is_finite() && weight > 0.0 check) filters
            // invalid weights before they reach here; this assert guards against regressions.
            debug_assert!(
                weight.is_finite() && weight > 0.0,
                "weighted_add called with invalid weight: {weight}"
            );
            if let Some(w) = map.get_mut(key) {
                *w += weight;
            } else {
                map.insert(key.to_vec(), weight);
            }
        }

        // Pre-compute function pointer for the hot path to avoid branching in the inner loop.
        // Uses borrowed-key lookup to avoid allocating Vec<u8> when key exists —
        // only allocates for new keys, eliminating most allocations for
        // low-cardinality columns.
        // NOTE: field_buffer is borrowed transiently by weighted_add (borrow ends within
        // the call), so reusing it across iterations is safe.
        let process_field: fn(&[u8], &mut HashMap<Vec<u8>, f64>, f64, &mut String, &mut Vec<u8>) =
            if flag_ignore_case {
                if flag_no_trim {
                    |field, map, weight, string_buf, field_buffer| {
                        if let Ok(s) = simdutf8::basic::from_utf8(field) {
                            util::to_lowercase_into(s, string_buf);
                            field_buffer.clear();
                            field_buffer.extend_from_slice(string_buf.as_bytes());
                            weighted_add(map, field_buffer, weight);
                        } else {
                            cold_path();
                            weighted_add(map, field, weight);
                        }
                    }
                } else {
                    // ignore-case + trim: use str::trim() for Unicode-aware whitespace trimming
                    |field, map, weight, string_buf, field_buffer| {
                        if let Ok(s) = simdutf8::basic::from_utf8(field) {
                            util::to_lowercase_into(s.trim(), string_buf);
                            field_buffer.clear();
                            field_buffer.extend_from_slice(string_buf.as_bytes());
                            weighted_add(map, field_buffer, weight);
                        } else {
                            cold_path();
                            weighted_add(map, trim_bs_whitespace(field), weight);
                        }
                    }
                }
            } else if flag_no_trim {
                |field, map, weight, _string_buf, _field_buffer| {
                    weighted_add(map, field, weight);
                }
            } else {
                // this is the default hot path
                |field, map, weight, _string_buf, _field_buffer| {
                    weighted_add(map, trim_bs_whitespace(field), weight);
                }
            };

        let mut row_result: csv::ByteRecord;
        for row in it {
            // safety: we know the row is valid because it comes from an iterator
            row_result = unsafe { row.unwrap_unchecked() };
            row_buffer.clone_from(&row_result);

            let weight = if let Some(widx) = weight_col_idx {
                if widx < row_result.len() {
                    // safety: widx < row_result.len() is checked above, so get(widx)
                    // will always return Some(&[u8])
                    fast_float2::parse::<f64, &[u8]>(row_result.get(widx).unwrap()).unwrap_or(1.0)
                } else {
                    1.0
                }
            } else {
                1.0
            };

            if !weight.is_finite() || weight <= 0.0 {
                continue;
            }

            for (i, field) in sel.select(&row_buffer).enumerate() {
                // safety: weighted_freq_tables is pre-allocated with sel_len elements.
                // i will always be < sel_len as it comes from enumerate() over the selected cols
                if !field.is_empty() {
                    process_field(
                        field,
                        unsafe { weighted_freq_tables.get_unchecked_mut(i) },
                        weight,
                        &mut string_buf,
                        &mut field_buffer,
                    );
                } else if !flag_no_nulls {
                    let map = unsafe { weighted_freq_tables.get_unchecked_mut(i) };
                    weighted_add(map, &[], weight);
                }
            }
        }

        weighted_freq_tables
    }

    #[inline]
    fn ftables_unweighted<I>(&self, sel: &Selection, it: I, nchunks: usize) -> FTables
    where
        I: Iterator<Item = csv::Result<csv::ByteRecord>>,
    {
        let sel_len = sel.len();

        let mut field_buffer: Vec<u8> = Vec::with_capacity(1024);
        let mut row_buffer: csv::ByteRecord = csv::ByteRecord::with_capacity(200, sel_len);
        let mut string_buf = String::with_capacity(512);

        let unique_headers_vec = UNIQUE_COLUMNS_VEC.get().unwrap();

        // assign flags to local variables for faster access
        let flag_no_nulls = self.flag_no_nulls;
        let flag_ignore_case = self.flag_ignore_case;
        let flag_no_trim = self.flag_no_trim;

        // compile a vector of bool flags for columns to skip in the hot loop.
        // Includes ALL_UNIQUE columns and columns with pre-built FTables from
        // the frequency cache (partial cache hit — merged back in run()).
        let mut all_unique_flag_vec: Vec<bool> = (0..sel_len)
            .map(|i| unique_headers_vec.contains(&i))
            .collect();
        if let Some(cache_skip) = FREQ_CACHE_SKIP.get() {
            for (i, &is_cached) in cache_skip.iter().enumerate() {
                if is_cached && i < sel_len {
                    all_unique_flag_vec[i] = true;
                }
            }
        }

        // optimize the capacity of the freq_tables based on the cardinality of the columns
        // if sequential, use the cardinality from the stats cache
        // if parallel, use a default capacity of 1000 for non-unique columns
        let col_cardinality_vec = COL_CARDINALITY_VEC.get().unwrap_or(&EMPTY_VEC);
        let mut freq_tables: Vec<_> = if col_cardinality_vec.is_empty() {
            (0..sel_len)
                .map(|_| Frequencies::with_capacity(1000))
                .collect()
        } else {
            (0..sel_len)
                .map(|i| {
                    let capacity = if all_unique_flag_vec[i] {
                        1
                    } else if nchunks == 1 {
                        col_cardinality_vec
                            .get(i)
                            .map_or(1000, |&cardinality| cardinality as usize)
                    } else {
                        // use cardinality and number of jobs to set the capacity
                        let cardinality = col_cardinality_vec
                            .get(i)
                            .map_or(1000, |&cardinality| cardinality as usize);
                        cardinality / nchunks
                    };
                    Frequencies::with_capacity(capacity)
                })
                .collect()
        };

        // Pre-compute function pointer for the hot path to avoid branching in the inner loop.
        // Uses add_borrowed() to avoid allocating a Vec<u8> for every field —
        // only the ignore-case path needs field_buffer; other paths pass &[u8] slices
        // directly, eliminating ~99% of allocations for low-cardinality columns.
        // NOTE: field_buffer is borrowed transiently by add_borrowed (borrow ends within
        // the call), so reusing it across iterations is safe.
        let process_field: fn(&[u8], &mut Frequencies<Vec<u8>>, &mut String, &mut Vec<u8>) =
            if flag_ignore_case {
                if flag_no_trim {
                    |field, ftab, string_buf, field_buffer| {
                        if let Ok(s) = simdutf8::basic::from_utf8(field) {
                            util::to_lowercase_into(s, string_buf);
                            field_buffer.clear();
                            field_buffer.extend_from_slice(string_buf.as_bytes());
                            ftab.add_borrowed(field_buffer);
                        } else {
                            cold_path();
                            ftab.add_borrowed(field);
                        }
                    }
                } else {
                    // ignore-case + trim: use str::trim() for Unicode-aware whitespace trimming
                    |field, ftab, string_buf, field_buffer| {
                        if let Ok(s) = simdutf8::basic::from_utf8(field) {
                            util::to_lowercase_into(s.trim(), string_buf);
                            field_buffer.clear();
                            field_buffer.extend_from_slice(string_buf.as_bytes());
                            ftab.add_borrowed(field_buffer);
                        } else {
                            cold_path();
                            ftab.add_borrowed(trim_bs_whitespace(field));
                        }
                    }
                }
            } else if flag_no_trim {
                |field, ftab, _string_buf, _field_buffer| {
                    ftab.add_borrowed(field);
                }
            } else {
                // this is the default hot path
                |field, ftab, _string_buf, _field_buffer| {
                    ftab.add_borrowed(trim_bs_whitespace(field));
                }
            };

        for row in it {
            // safety: we know the row is valid
            row_buffer.clone_from(&unsafe { row.unwrap_unchecked() });
            for (i, field) in sel.select(&row_buffer).enumerate() {
                // safety: all_unique_flag_vec is pre-computed to have exactly sel_len elements,
                // which matches the number of selected columns that we iterate over.
                // i will always be < sel_len as it comes from enumerate() over the selected cols
                if unsafe { *all_unique_flag_vec.get_unchecked(i) } {
                    continue;
                }

                // safety: freq_tables is pre-allocated with sel_len elements.
                // i will always be < sel_len as it comes from enumerate() over the selected cols
                if !field.is_empty() {
                    process_field(
                        field,
                        unsafe { freq_tables.get_unchecked_mut(i) },
                        &mut string_buf,
                        &mut field_buffer,
                    );
                } else if !flag_no_nulls {
                    // set to null (EMPTY_BYTES) as flag_no_nulls is false
                    unsafe {
                        freq_tables.get_unchecked_mut(i).add_borrowed(&[]);
                    }
                }
            }
        }
        // shrink the capacity of the freq_tables to the actual number of elements.
        // if sequential (nchunks == 1), we don't need to shrink the capacity as we
        // use cardinality to set the capacity of the freq_tables
        // if parallel (nchunks > 1), shrink the capacity to avoid over-allocating memory
        if nchunks > 1 {
            freq_tables.shrink_to_fit();
        }
        freq_tables
    }

    /// Compute indices of Float columns that should be skipped from frequency analysis.
    ///
    /// # Arguments
    /// * `headers` - The selected CSV headers (used for the --no-float exception list, which
    ///   matches by user-supplied column name)
    /// * `sel` - The selection; `sel[i]` is the ORIGINAL CSV column position of selected column
    ///   `i`, used to look up the column type positionally
    /// * `col_type_by_pos` - Column types indexed by ORIGINAL CSV column position
    ///
    /// # Returns
    /// A vector of ORIGINAL CSV column indices of Float columns to skip. Original indices
    /// (not selected positions) are stable when an earlier filter removes columns, so this
    /// composes correctly with --stats-filter.
    /// Columns listed in the --no-float exception list are NOT included in the skip list.
    fn compute_float_columns_to_skip(
        &self,
        headers: &Headers,
        sel: &Selection,
        col_type_by_pos: &[String],
    ) -> Vec<usize> {
        // Parse exception columns from flag value
        // If value is "*", exclude ALL Float columns (empty exception list)
        // Otherwise, parse as comma-separated list of Float columns to INCLUDE
        let exception_cols: HashSet<String> = self
            .flag_no_float
            .as_ref()
            .map(|cols| {
                let trimmed = cols.trim();
                // "*" means exclude all floats (no exceptions)
                if trimmed == "*" {
                    HashSet::new()
                } else {
                    trimmed
                        .split(',')
                        .map(|s| s.trim().to_lowercase())
                        .filter(|s| !s.is_empty() && s != "*")
                        .collect()
                }
            })
            .unwrap_or_default();

        let mut float_columns_to_skip = Vec::new();

        // Look up the type by ORIGINAL column position (sel[i]) — not by header name — so
        // duplicate-named columns are each evaluated on their own type. The exception list
        // still matches by name (it's a user-supplied list of column names). We record the
        // ORIGINAL column index (orig_idx), which stays valid even if another filter later
        // removes columns ahead of it.
        for (header, &orig_idx) in headers.iter().zip(sel.iter()) {
            if let Some(col_type) = col_type_by_pos.get(orig_idx)
                && col_type == "Float"
            {
                let header_str = simdutf8::basic::from_utf8(header)
                    .unwrap_or(NON_UTF8_ERR)
                    .to_lowercase();
                if !exception_cols.contains(&header_str) {
                    float_columns_to_skip.push(orig_idx);
                }
            }
        }

        float_columns_to_skip
    }

    /// Compute indices of columns that should be skipped based on the --stats-filter expression.
    ///
    /// # Arguments
    /// * `headers` - The selected CSV headers (used only for log/error messages)
    /// * `sel` - The selection; `sel[i]` is the ORIGINAL CSV column position of selected column
    ///   `i`, used to look up the stats record positionally
    /// * `stats_by_pos` - Stats records indexed by ORIGINAL CSV column position
    /// * `filter_expression` - The Luau expression to evaluate for each column
    ///
    /// # Returns
    /// A vector of ORIGINAL CSV column indices of columns where the filter expression
    /// evaluated to `true` (meaning they should be excluded). Original indices (not selected
    /// positions) stay valid even if --no-float removed columns first, so the two filters
    /// compose correctly.
    #[allow(clippy::unused_self)]
    #[cfg(feature = "luau")]
    fn compute_stats_filter_columns_to_skip(
        &self,
        headers: &Headers,
        sel: &Selection,
        stats_by_pos: &[StatsData],
        filter_expression: &str,
    ) -> CliResult<Vec<usize>> {
        use mlua::Lua;

        // Create a single sandboxed Luau VM for all column evaluations
        let lua = Lua::new();
        lua.sandbox(true)
            .map_err(|e| format!("Failed to enable Luau sandbox: {e}"))?;

        let mut columns_to_skip = Vec::new();

        // Look up the stats record by ORIGINAL column position (sel[i]) — not by header
        // name — so duplicate-named columns are each evaluated against their own stats. We
        // record the ORIGINAL column index (orig_idx), which stays valid even if --no-float
        // removed columns ahead of it.
        for (header, &orig_idx) in headers.iter().zip(sel.iter()) {
            let Some(stats_data) = stats_by_pos.get(orig_idx) else {
                // No stats available for this column - skip filtering for it
                log::debug!(
                    "No stats available for column index {orig_idx}, skipping --stats-filter \
                     evaluation"
                );
                continue;
            };

            // Evaluate the filter expression against this column's stats
            match evaluate_stats_filter(&lua, stats_data, filter_expression) {
                Ok(should_exclude) => {
                    if should_exclude {
                        let header_str = simdutf8::basic::from_utf8(header).unwrap_or(NON_UTF8_ERR);
                        log::debug!("Column '{header_str}' excluded by --stats-filter expression");
                        columns_to_skip.push(orig_idx);
                    }
                },
                Err(e) => {
                    let header_str = simdutf8::basic::from_utf8(header).unwrap_or(NON_UTF8_ERR);
                    return fail_clierror!(
                        "Error evaluating --stats-filter expression for column '{header_str}': {e}"
                    );
                },
            }
        }

        Ok(columns_to_skip)
    }

    /// Return the all-unique columns (cardinality == rowcount) as ORIGINAL CSV
    /// column indices, together with a cardinality vector indexed by ORIGINAL CSV
    /// column position (aligned to the stats cache, which has one record per
    /// column regardless of duplicate names).
    ///
    /// Both are keyed by position — NOT by header name — so duplicate-named columns
    /// are each classified on their own data. `sel` maps each selected column to its
    /// original CSV column position. An empty cardinality vector signals "no stats
    /// cache available" (compute frequencies for all columns).
    ///
    /// Also stores the stats records in a hashmap for use when producing JSON output
    /// and computes the Float / `--stats-filter` skip lists (both still name-keyed,
    /// as those features index by selected-header name).
    fn get_unique_headers(
        &self,
        sel: &Selection,
        headers: &Headers,
    ) -> CliResult<(Vec<usize>, Vec<u64>, Vec<StatsData>)> {
        // get the stats records for the entire CSV
        let schema_args = util::SchemaArgs {
            flag_enum_threshold:  0,
            flag_ignore_case:     self.flag_ignore_case,
            flag_strict_dates:    false,
            flag_strict_formats:  false,
            // we still get all the stats columns so we can use the stats cache
            flag_pattern_columns: crate::select::SelectColumns::parse("").unwrap(),
            flag_dates_whitelist: String::new(),
            flag_prefer_dmy:      false,
            flag_force:           false,
            flag_stdout:          false,
            flag_jobs:            Some(util::njobs(self.flag_jobs)),
            flag_polars:          false,
            flag_no_headers:      self.flag_no_headers,
            flag_delimiter:       self.flag_delimiter,
            arg_input:            self.arg_input.clone(),
            flag_memcheck:        false,
            flag_output:          None,
        };
        let is_json = self.flag_json || self.flag_pretty_json || self.flag_toon;

        // What positional/name-keyed structures each feature needs, so we only pay for
        // what's requested. ALL per-column lookups are positional (indexed by ORIGINAL CSV
        // column position) so duplicate-named columns are each classified on their own data
        // — the lone exception is the name-keyed hashmap below, which exists solely for the
        // --weight tolerance lookup (it addresses ONE column by the user-supplied name).
        let needs_weight_records = is_json && self.flag_weight.is_some();
        let needs_float_types = self.flag_no_float.is_some();
        #[cfg(feature = "luau")]
        let needs_stats_by_pos = is_json || self.flag_stats_filter.is_some();
        #[cfg(not(feature = "luau"))]
        let needs_stats_by_pos = is_json;

        // initialize the name-keyed stats records hashmap (used only by --weight tolerance)
        let mut stats_records_hashmap = if needs_weight_records {
            HashMap::with_capacity(headers.len())
        } else {
            HashMap::new()
        };

        let (csv_fields, csv_stats) = get_stats_records(&schema_args, StatsMode::Frequency)?;

        if csv_fields.is_empty() || csv_stats.len() != csv_fields.len() {
            // the stats cache does not exist or the number of fields & stats records
            // do not match. Just return empty vectors.
            // we're not going to be able to get the cardinalities, so
            // this signals that we just compute frequencies for all columns
            return Ok((Vec::new(), Vec::new(), Vec::new()));
        }

        // Positional vectors indexed by ORIGINAL CSV column position. csv_stats/csv_fields
        // have one entry per column in original order, so the index IS the original column
        // position — duplicate-named columns each keep their own cardinality/type/stats (a
        // name-keyed lookup would collapse them onto the first match).
        let mut col_cardinality_by_pos: Vec<u64> = Vec::with_capacity(csv_stats.len());
        // Per-column types for --no-float detection (built only when needed).
        let mut col_type_by_pos: Vec<String> = if needs_float_types {
            Vec::with_capacity(csv_stats.len())
        } else {
            Vec::new()
        };
        // Per-column stats for JSON/TOON output and --stats-filter, aligned to ORIGINAL
        // columns. For JSON, the caller (sel_headers) remaps it to the final selected order.
        let mut stats_by_pos: Vec<StatsData> = if needs_stats_by_pos {
            Vec::with_capacity(csv_stats.len())
        } else {
            Vec::new()
        };

        for (i, stats_record) in csv_stats.iter().enumerate() {
            if needs_weight_records {
                // Name-keyed record for the --weight tolerance lookup (addresses the weight
                // column by user-supplied name). safety: csv_fields and csv_stats are equal
                // length, so index i is valid for csv_fields.
                let col_name = csv_fields.get(i).unwrap();
                let col_name_str = simdutf8::basic::from_utf8(col_name)
                    .unwrap_or(NON_UTF8_ERR)
                    .to_string();
                stats_records_hashmap.insert(col_name_str, stats_record.clone());
            }
            if needs_float_types {
                col_type_by_pos.push(stats_record.r#type.clone());
            }
            if needs_stats_by_pos {
                stats_by_pos.push(stats_record.clone());
            }
            col_cardinality_by_pos.push(stats_record.cardinality);
        }

        // now, get the unique headers, where cardinality == rowcount
        let row_count = util::count_rows(&self.rconfig()).unwrap_or_default();
        FREQ_ROW_COUNT.set(row_count).unwrap();

        // Determine all-unique columns by ORIGINAL column position via the selection,
        // not by header name, so duplicate-named columns are each classified on their
        // own cardinality. `sel` yields the original CSV column index of each selected
        // column. Returns original indices (the caller remaps them to selected positions).
        // Most datasets have relatively few all-unique columns (e.g. ID columns), so
        // pre-allocate space for 5 as a reasonable default capacity.
        let mut all_unique_headers_vec: Vec<usize> = Vec::with_capacity(5);
        for &orig_idx in sel.iter() {
            let cardinality = col_cardinality_by_pos.get(orig_idx).copied().unwrap_or(0);
            if cardinality == row_count {
                all_unique_headers_vec.push(orig_idx);
            }
        }

        // Compute Float columns to skip if --no-float is specified
        if self.flag_no_float.is_some() {
            let float_columns_to_skip =
                self.compute_float_columns_to_skip(headers, sel, &col_type_by_pos);
            if FLOAT_COLUMNS_TO_SKIP.set(float_columns_to_skip).is_err() {
                log::warn!("FLOAT_COLUMNS_TO_SKIP already set — stale float columns may be used");
            }
        }

        // Compute stats filter columns to skip if --stats-filter is specified
        #[cfg(feature = "luau")]
        if let Some(ref filter_expression) = self.flag_stats_filter {
            if stats_by_pos.is_empty() {
                log::warn!(
                    "Stats cache unavailable. Cannot apply --stats-filter. Run 'qsv stats \
                     --cardinality --stats-jsonl' first."
                );
            } else {
                let stats_filter_columns_to_skip = self.compute_stats_filter_columns_to_skip(
                    headers,
                    sel,
                    &stats_by_pos,
                    filter_expression,
                )?;
                if STATS_FILTER_COLUMNS_TO_SKIP
                    .set(stats_filter_columns_to_skip)
                    .is_err()
                {
                    log::warn!(
                        "STATS_FILTER_COLUMNS_TO_SKIP already set — stale filter columns may be \
                         used"
                    );
                }
            }
        }

        if is_json && self.flag_weight.is_some() {
            // Only the weighted-JSON tolerance path reads STATS_RECORDS; set it just for
            // that case (addresses the weight column by name). Non-weighted JSON never
            // populated the hashmap above, so there is nothing to store.
            STATS_RECORDS.set(stats_records_hashmap).unwrap();
        }

        // COL_CARDINALITY_VEC and STATS_RECORDS_BY_POS are set by the caller (sel_headers)
        // once the FINAL selection is known, so they can be aligned positionally to the
        // final columns.
        Ok((all_unique_headers_vec, col_cardinality_by_pos, stats_by_pos))
    }

    #[allow(clippy::cast_precision_loss)]
    fn output_json(
        &self,
        headers: &Headers,
        tables: FTables,
        weighted_tables: Option<&WeightedFTables>,
        rconfig: &Config,
        argv: &[&str],
        is_stdin: bool,
    ) -> CliResult<()> {
        let fieldcount = headers.len();

        // init vars and amortize allocations
        let mut fields = Vec::with_capacity(fieldcount);
        let rowcount = *FREQ_ROW_COUNT.get().unwrap_or(&0);
        let unique_headers_vec = UNIQUE_COLUMNS_VEC.get().unwrap();
        let mut processed_frequencies = Vec::with_capacity(headers.len());
        let abs_dec_places = self.flag_pct_dec_places.unsigned_abs() as u32;
        // pre-allocate space for 17 field stats, see list below for details
        let mut field_stats: Vec<FieldStats> = Vec::with_capacity(17);

        // Helper function to build a frequency field for JSON output
        let build_frequency_field = |field_name: String,
                                     field_pos: usize,
                                     cardinality: u64,
                                     processed_frequencies: &mut Vec<ProcessedFrequency>,
                                     field_stats: &mut Vec<FieldStats>,
                                     skip_stats: bool| {
            // Sort frequencies by count if flag_other_sorted,
            // breaking ties by value for deterministic output
            if self.flag_other_sorted {
                if self.flag_asc {
                    processed_frequencies.sort_unstable_by(|a, b| {
                        a.count.cmp(&b.count).then_with(|| a.value.cmp(&b.value))
                    });
                } else {
                    processed_frequencies.sort_unstable_by(|a, b| {
                        b.count.cmp(&a.count).then_with(|| a.value.cmp(&b.value))
                    });
                }
            }

            // Get stats record for this field by output-column POSITION (not name), so
            // duplicate-named columns each get their own stats.
            let stats_record = STATS_RECORDS_BY_POS
                .get()
                .and_then(|records| records.get(field_pos));

            // Get data type and nullcount from stats record
            let dtype = stats_record.map_or(String::new(), |sr| sr.r#type.clone());
            let nullcount = stats_record.map_or(0, |sr| sr.nullcount);
            let sparsity =
                fast_float2::parse(util::round_num(nullcount as f64 / rowcount as f64, 4))
                    .unwrap_or(0.0);
            let uniqueness_ratio =
                fast_float2::parse(util::round_num(cardinality as f64 / rowcount as f64, 4))
                    .unwrap_or(0.0);

            // Build stats vector from stats record if type is not empty and not NULL or Boolean
            // Skip stats when using weighted mode (stats would be misleading)
            if !self.flag_no_stats
                && !skip_stats
                && !dtype.is_empty()
                && dtype.as_str() != "NULL"
                && dtype.as_str() != "Boolean"
                && let Some(sr) = stats_record
            {
                // Add all available stats if some
                add_stat(field_stats, "sum", sr.sum);
                add_stat(field_stats, "min", sr.min.clone());
                add_stat(field_stats, "max", sr.max.clone());
                add_stat(field_stats, "range", sr.range);
                add_stat(field_stats, "sort_order", sr.sort_order.clone());

                // String-specific length stats
                add_stat(field_stats, "min_length", sr.min_length);
                add_stat(field_stats, "max_length", sr.max_length);
                add_stat(field_stats, "sum_length", sr.sum_length);
                add_stat(field_stats, "avg_length", sr.avg_length);
                add_stat(field_stats, "stddev_length", sr.stddev_length);
                add_stat(field_stats, "variance_length", sr.variance_length);
                add_stat(field_stats, "cv_length", sr.cv_length);

                // Numeric-specific stats
                add_stat(field_stats, "mean", sr.mean);
                add_stat(field_stats, "sem", sr.sem);
                add_stat(field_stats, "stddev", sr.stddev);
                add_stat(field_stats, "variance", sr.variance);
                add_stat(field_stats, "cv", sr.cv);
            }

            FrequencyField {
                field: field_name,
                r#type: dtype,
                cardinality,
                nullcount,
                sparsity,
                uniqueness_ratio,
                stats: std::mem::take(field_stats),
                frequencies: processed_frequencies
                    .iter()
                    .map(|pf| {
                        // Sentinel value -1.0 indicates NULL entry with --pct-nulls=false
                        let (pct_opt, rank_opt) = if pf.percentage < 0.0 {
                            (None, None)
                        } else {
                            (
                                Some(
                                    fast_float2::parse(&pf.formatted_percentage)
                                        .unwrap_or(pf.percentage),
                                ),
                                Some(pf.rank),
                            )
                        };
                        FrequencyEntry {
                            value:      if self.flag_vis_whitespace {
                                util::visualize_whitespace(&util::bytes_to_cow_str(&pf.value))
                            } else {
                                util::bytes_to_cow_str(&pf.value).into_owned()
                            },
                            count:      pf.count,
                            percentage: pct_opt,
                            rank:       rank_opt,
                        }
                    })
                    .collect(),
            }
        };

        if let Some(weighted) = weighted_tables {
            // Process weighted frequencies for JSON output
            for (i, header) in headers.iter().enumerate() {
                let field_name = if rconfig.no_headers {
                    (i + 1).to_string()
                } else {
                    util::bytes_to_cow_str(header).into_owned()
                };

                let all_unique_header = unique_headers_vec.contains(&i);
                if i < weighted.len() {
                    self.process_frequencies_weighted(
                        all_unique_header,
                        abs_dec_places,
                        rowcount,
                        &weighted[i],
                        &mut processed_frequencies,
                    );
                }

                // Calculate cardinality for this field
                let cardinality = if all_unique_header {
                    rowcount
                } else if i < weighted.len() {
                    weighted[i].len() as u64
                } else {
                    0
                };

                fields.push(build_frequency_field(
                    field_name,
                    i,
                    cardinality,
                    &mut processed_frequencies,
                    &mut field_stats,
                    true, // Skip stats for weighted mode
                ));

                processed_frequencies.clear(); // clear for next field
            }
        } else {
            // Process unweighted frequencies for JSON output
            let head_ftables = headers.iter().zip(tables);
            for (i, (header, ftab)) in head_ftables.enumerate() {
                let field_name = if rconfig.no_headers {
                    (i + 1).to_string()
                } else {
                    util::bytes_to_cow_str(header).into_owned()
                };

                let all_unique_header = unique_headers_vec.contains(&i);
                self.process_frequencies(
                    all_unique_header,
                    abs_dec_places,
                    rowcount,
                    &ftab,
                    &mut processed_frequencies,
                );

                // Calculate cardinality for this field
                let cardinality = if all_unique_header {
                    rowcount // For all-unique fields, cardinality == rowcount
                } else {
                    ftab.len() as u64 // otherwise, cardinality == number of unique values
                };

                fields.push(build_frequency_field(
                    field_name,
                    i,
                    cardinality,
                    &mut processed_frequencies,
                    &mut field_stats,
                    false, // Include stats for non-weighted mode
                ));

                processed_frequencies.clear(); // clear for next field
            }
        }

        let output = FrequencyOutput {
            input: if is_stdin {
                "stdin".to_string()
            } else {
                // safety: we know arg_input is not None
                self.arg_input.clone().unwrap()
            },
            description: format!("Generated with `qsv {}`", argv[1..].join(" ")),
            rowcount: if rowcount == 0 {
                // if rowcount == 0, derive the rowcount from first field's frequencies
                // by summing the counts for the first field
                fields
                    .first()
                    .map_or(0, |field| field.frequencies.iter().map(|f| f.count).sum())
            } else {
                rowcount
            },
            fieldcount,
            fields,
            rank_strategy: self.flag_rank_strategy,
        };

        if self.flag_toon {
            // TOON output - encode the JSON structure to TOON format.
            // Empty `stats` arrays are skipped by serde (#[serde(skip_serializing_if)]),
            // so no manual post-processing is needed.
            let json_value = serde_json::to_value(&output)?;
            let opts = EncodeOptions::new();
            let toon_output = encode(&json_value, &opts)
                .map_err(|e| crate::CliError::Other(format!("Failed to encode to TOON: {e}")))?;
            if let Some(output_path) = &self.flag_output {
                std::fs::write(output_path, toon_output)?;
            } else {
                println!("{toon_output}");
            }
        } else {
            // JSON output. Empty `stats` arrays are skipped by serde, so no
            // post-processing is needed.
            let json_output = if self.flag_pretty_json {
                // pretty, with more whitespace
                serde_json::to_string_pretty(&output)?
            } else {
                // still pretty, but more compact and faster
                simd_json::to_string_pretty(&output)?
            };

            if let Some(output_path) = &self.flag_output {
                std::fs::write(output_path, json_output)?;
            } else {
                println!("{json_output}");
            }
        }

        Ok(())
    }

    /// Processes headers and handles weight column exclusion if needed.
    ///
    /// This function handles the logic for excluding the weight column from frequency
    /// computation. It finds the weight column index, creates a modified selection that
    /// excludes it, and returns the selected headers.
    ///
    /// # Arguments
    ///
    /// * `full_headers` - The full CSV headers as a `ByteRecord`
    ///
    /// # Returns
    ///
    /// * `Ok((Option<usize>, Selection, csv::ByteRecord))` - Tuple containing:
    ///   - Weight column index (None if no weight column specified)
    ///   - Modified selection (excluding weight column if present)
    ///   - Selected headers (excluding weight column if present)
    /// * `Err(CliError)` - If weight column is not found or no columns remain after exclusion
    fn process_headers_with_weight_exclusion(
        &self,
        full_headers: &csv::ByteRecord,
    ) -> CliResult<(Option<usize>, Selection, csv::ByteRecord)> {
        if let Some(ref weight_col) = self.flag_weight {
            // Find weight column index in full headers
            let weight_idx = full_headers
                .iter()
                .position(|h| {
                    let h_str = util::bytes_to_cow_str(h);
                    h_str.trim().eq_ignore_ascii_case(weight_col.trim())
                })
                .ok_or_else(|| {
                    crate::CliError::Other(format!(
                        "Weight column '{weight_col}' not found in CSV headers"
                    ))
                })?;

            // Create selection excluding weight column
            let sel = self.rconfig().selection(full_headers)?;
            // Remove weight column index from selection if present
            let sel_vec: Vec<usize> = sel
                .iter()
                .copied()
                .filter(|&idx| idx != weight_idx)
                .collect();

            // Validate that we still have columns after excluding the weight column
            if sel_vec.is_empty() {
                return Err(crate::CliError::Other(format!(
                    "After excluding weight column '{weight_col}', no columns remain for \
                     frequency computation"
                )));
            }

            let modified_sel = Selection::from_indices(sel_vec);

            // Get selected headers (excluding weight column)
            let selected_headers: csv::ByteRecord = modified_sel.select(full_headers).collect();

            Ok((Some(weight_idx), modified_sel, selected_headers))
        } else {
            // No weight column specified, use normal selection
            let sel = self.rconfig().selection(full_headers)?;
            let headers: csv::ByteRecord = sel.select(full_headers).collect();
            Ok((None, sel, headers))
        }
    }

    fn sel_headers<R: io::Read>(
        &self,
        rdr: &mut csv::Reader<R>,
    ) -> CliResult<(csv::ByteRecord, Selection, Option<usize>)> {
        let full_headers = rdr.byte_headers()?.clone();
        let (weight_col_idx, mut sel, selected_headers) =
            self.process_headers_with_weight_exclusion(&full_headers)?;

        let (all_unique_headers_vec, col_cardinality_by_pos, stats_by_pos) =
            self.get_unique_headers(&sel, &selected_headers)?;

        // Filter out Float columns if --no-float is specified
        let (final_sel, final_headers) = if self.flag_no_float.is_some() {
            if let Some(float_cols_to_skip) = FLOAT_COLUMNS_TO_SKIP.get() {
                if float_cols_to_skip.is_empty() {
                    (sel, selected_headers)
                } else {
                    // Filter selection to exclude Float columns.
                    // float_cols_to_skip contains ORIGINAL CSV column indices, and sel
                    // yields original column indices, so filter by the index value (stable
                    // regardless of position).
                    let sel_vec: Vec<usize> = sel
                        .iter()
                        .copied()
                        .filter(|orig_idx| !float_cols_to_skip.contains(orig_idx))
                        .collect();

                    if sel_vec.is_empty() {
                        return Err(crate::CliError::Other(
                            "No columns remain after excluding Float columns. Use --no-float with \
                             exception columns to include specific Float columns."
                                .to_string(),
                        ));
                    }

                    sel = Selection::from_indices(sel_vec);
                    let headers: csv::ByteRecord = sel.select(&full_headers).collect();
                    (sel, headers)
                }
            } else {
                // Stats cache unavailable for Float type detection
                log::warn!(
                    "Stats cache unavailable. Cannot detect Float columns for --no-float. \
                     Processing all columns. Run 'qsv stats --cardinality --stats-jsonl' first."
                );
                (sel, selected_headers)
            }
        } else {
            (sel, selected_headers)
        };

        // Filter out stats-filtered columns if --stats-filter is specified
        #[cfg(feature = "luau")]
        let (final_sel, final_headers) = if self.flag_stats_filter.is_some() {
            if let Some(stats_filter_cols_to_skip) = STATS_FILTER_COLUMNS_TO_SKIP.get() {
                if stats_filter_cols_to_skip.is_empty() {
                    (final_sel, final_headers)
                } else {
                    // Filter selection to exclude stats-filtered columns.
                    // stats_filter_cols_to_skip contains ORIGINAL CSV column indices; since
                    // final_sel yields original column indices, this stays correct even when
                    // --no-float already dropped earlier columns (positions would have shifted).
                    let sel_vec: Vec<usize> = final_sel
                        .iter()
                        .copied()
                        .filter(|orig_idx| !stats_filter_cols_to_skip.contains(orig_idx))
                        .collect();

                    if sel_vec.is_empty() {
                        return Err(crate::CliError::Other(
                            "No columns remain after applying --stats-filter. Adjust your filter \
                             expression to be less restrictive."
                                .to_string(),
                        ));
                    }

                    let new_sel = Selection::from_indices(sel_vec);
                    let headers: csv::ByteRecord = new_sel.select(&full_headers).collect();
                    (new_sel, headers)
                }
            } else {
                // Stats cache unavailable for --stats-filter
                log::warn!(
                    "Stats cache unavailable. Cannot apply --stats-filter. Processing all \
                     columns. Run 'qsv stats --cardinality --stats-jsonl' first."
                );
                (final_sel, final_headers)
            }
        } else {
            (final_sel, final_headers)
        };

        // Map original column indices to selected column indices
        let mapped_unique_headers: Vec<usize> = all_unique_headers_vec
            .iter()
            .filter_map(|&original_idx| {
                // Find the position of this original index in the selection
                final_sel
                    .iter()
                    .position(|&sel_idx| sel_idx == original_idx)
            })
            .collect();

        UNIQUE_COLUMNS_VEC
            .set(mapped_unique_headers)
            .map_err(|_| "Cannot set UNIQUE_COLUMNS")?;

        // Build the cardinality vector aligned positionally to the FINAL selected
        // columns (after --no-float / --stats-filter dropping), so downstream
        // consumers (cache cardinality, capacity hints) can index by position.
        // Skip when the stats cache is unavailable (empty), leaving COL_CARDINALITY_VEC
        // unset so consumers fall back to default capacities.
        if !col_cardinality_by_pos.is_empty() {
            let final_cardinalities: Vec<u64> = final_sel
                .iter()
                .map(|&orig_idx| col_cardinality_by_pos.get(orig_idx).copied().unwrap_or(0))
                .collect();
            if COL_CARDINALITY_VEC.set(final_cardinalities).is_err() {
                log::warn!("COL_CARDINALITY_VEC already set — stale cardinalities may be used");
            }
        }

        // Likewise, align the per-output-column JSON stats to the FINAL selected columns
        // (positional), so duplicate-named columns each report their own stats. orig_idx
        // values come from a valid selection over the original columns, so every lookup
        // resolves; the length guard skips the set if any unexpectedly does not.
        // Only for JSON output: stats_by_pos may also be populated for --stats-filter in
        // non-JSON mode, where STATS_RECORDS_BY_POS is never read.
        let is_json = self.flag_json || self.flag_pretty_json || self.flag_toon;
        if is_json && !stats_by_pos.is_empty() {
            let final_stats: Vec<StatsData> = final_sel
                .iter()
                .filter_map(|&orig_idx| stats_by_pos.get(orig_idx).cloned())
                .collect();
            if final_stats.len() == final_sel.len()
                && STATS_RECORDS_BY_POS.set(final_stats).is_err()
            {
                log::warn!("STATS_RECORDS_BY_POS already set — stale stats may be used");
            }
        }

        Ok((final_headers, final_sel, weight_col_idx))
    }

    /// Self-contained Frequent Items (Misra-Gries) processing path.
    ///
    /// Streams the input CSV once, maintaining one
    /// `datasketches::frequencies::FrequentItemsSketch<Vec<u8>>` per selected column,
    /// then emits the same CSV schema as the exact path
    /// (`field,value,count,percentage,rank`) using the sketch's heavy-hitter list.
    ///
    /// This path is intentionally narrow:
    ///   * Always unweighted, single-threaded — `run()` rejects `--weight` and the parallel/index
    ///     dispatch is bypassed (FI sketches merge cleanly via `FrequentItemsSketch::merge`, but
    ///     the MVP keeps everything sequential to limit blast radius).
    ///   * No frequency cache — caller skips `try_output_from_cache` since the cache stores exact
    ///     `Frequencies` state, not sketch state.
    ///   * Unicode trim is the only normalization applied (matches the default exact path);
    ///     `--ignore-case` and `--no-trim` are rejected upstream.
    ///   * Counts are estimates. The `count` column carries the FI point estimate; its error is
    ///     bounded by `sketch.maximum_error()` (which equals the stream length minus the active
    ///     threshold). The "Other" row's count is `total_weight - sum(top_k_estimates)` and is
    ///     therefore approximate.
    #[cfg(not(target_endian = "big"))]
    #[allow(clippy::cast_precision_loss)]
    fn run_frequent_items(&self, rconfig: &Config) -> CliResult<()> {
        use datasketches::frequencies::{ErrorType, FrequentItemsSketch};

        let mut rdr = rconfig.reader()?;
        let (headers, sel, weight_col_idx) = self.sel_headers(&mut rdr)?;
        debug_assert!(
            weight_col_idx.is_none(),
            "run_frequent_items reached with a weight column — run() should reject --weight"
        );

        let sel_len = sel.len();
        let map_size = self.flag_sketch_map_size;
        let mut sketches: Vec<FrequentItemsSketch<Vec<u8>>> = (0..sel_len)
            .map(|_| FrequentItemsSketch::<Vec<u8>>::new(map_size))
            .collect();

        // Per-column non-null observation count, used as the percentage denominator
        // when --pct-nulls is false (the default — null cells are excluded from the
        // valid-percentage base). We track it ourselves rather than calling
        // sketch.total_weight() because total_weight folds nulls in.
        let mut nonnull_counts: Vec<u64> = vec![0; sel_len];
        // Per-column null observation count (only populated when nulls are tracked).
        let mut null_counts: Vec<u64> = vec![0; sel_len];

        let flag_no_nulls = self.flag_no_nulls;
        // Reuse the same default-resolution path as the exact code: --null-text has
        // its docopt default of "(NULL)", so self.flag_null_text is always populated
        // by the time run() dispatches here. Reading it directly avoids depending
        // on NULL_VAL.set() ordering (the FI dispatch happens before that init in
        // run()) and guarantees exact↔FI label parity.
        //
        // Invariant: the only `NULL_VAL.set(...)` site in this module is at the top
        // of `run()` (a single setter — verifiable by grepping `NULL_VAL\.set` in
        // this file), and it sets `NULL_VAL` from
        // `args.flag_null_text.as_bytes().to_vec()` — the same bytes we read here.
        // So `NULL_VAL` and `self.flag_null_text` are byte-identical once
        // `NULL_VAL` is populated. If a future change adds another `NULL_VAL.set`
        // site that uses a different source, revisit this comment.
        let null_label: Vec<u8> = self.flag_null_text.as_bytes().to_vec();

        let mut row_buffer: csv::ByteRecord = csv::ByteRecord::with_capacity(200, sel_len);
        for row in rdr.byte_records() {
            row_buffer.clone_from(&row?);
            for (i, field) in sel.select(&row_buffer).enumerate() {
                if field.is_empty() {
                    if !flag_no_nulls {
                        // Track empty cells as a sentinel value so they can show up as
                        // a heavy hitter row labelled with --null-text. Use the empty
                        // Vec<u8> as the sketch key — distinct from any real value.
                        sketches[i].update(Vec::new());
                        null_counts[i] += 1;
                    }
                    continue;
                }
                // Default trim (ignore-case + no-trim are rejected upstream).
                let trimmed = trim_bs_whitespace(field);
                if trimmed.is_empty() {
                    if !flag_no_nulls {
                        sketches[i].update(Vec::new());
                        null_counts[i] += 1;
                    }
                    continue;
                }
                sketches[i].update(trimmed.to_vec());
                nonnull_counts[i] += 1;
            }
        }

        // Build the writer and emit.
        let mut wtr = Config::new(self.flag_output.as_ref()).writer()?;
        wtr.write_record([
            b"field".as_slice(),
            b"value",
            b"count",
            b"percentage",
            b"rank",
        ])?;

        // For percentage, mirror the default exact behavior: --pct-nulls=false uses
        // the non-null count as the denominator (so the null row's percentage is
        // empty), --pct-nulls=true folds nulls in.
        let pct_nulls = self.flag_pct_nulls;
        let abs_dec_places = self.flag_pct_dec_places.unsigned_abs() as u32;

        // --limit semantics:
        //   limit > 0   → keep first `limit` rows (top-K)
        //   limit == 0  → keep everything the sketch returns
        //   limit < 0   → keep rows with estimate >= |limit|
        let limit = self.flag_limit;

        // --no-other / --other-text "<NONE>" controls Other emission.
        let emit_other = !self.flag_no_other && self.flag_other_text != "<NONE>";
        let other_label = self.flag_other_text.as_bytes().to_vec();

        let mut itoa_buf = itoa::Buffer::new();
        let mut rank_buf = String::with_capacity(20);
        let mut value_str = String::with_capacity(64);

        let vis_whitespace = self.flag_vis_whitespace;
        let no_headers = rconfig.no_headers;

        for (i, header) in headers.iter().enumerate() {
            let header_vec: Vec<u8> = if no_headers {
                (i + 1).to_string().into_bytes()
            } else {
                header.to_vec()
            };
            let sketch = &sketches[i];
            let nonnull_count = nonnull_counts[i];
            let null_count = null_counts[i];
            let pct_denom: f64 = if pct_nulls {
                (nonnull_count + null_count) as f64
            } else {
                nonnull_count as f64
            };

            // NoFalsePositives ensures every emitted item's true count is >= lower_bound,
            // so we never report a value that wasn't actually a heavy hitter.
            let mut rows = sketch.frequent_items(ErrorType::NoFalsePositives);
            // Already sorted by estimate descending.
            // Apply --limit semantics on top of the sketch's natural ordering.
            if limit > 0 {
                rows.truncate(limit as usize);
            } else if limit < 0 {
                let min_count = limit.unsigned_abs() as u64;
                rows.retain(|r| r.estimate() >= min_count);
            }

            let kept_estimate_sum: u64 = rows
                .iter()
                .map(datasketches::frequencies::Row::estimate)
                .sum();

            for (rank_idx, r) in rows.iter().enumerate() {
                let item: &[u8] = r.item();
                let display_bytes: Vec<u8> = if item.is_empty() {
                    null_label.clone()
                } else if vis_whitespace {
                    value_str.clear();
                    value_str.push_str(&util::visualize_whitespace(&util::bytes_to_cow_str(item)));
                    value_str.as_bytes().to_vec()
                } else {
                    item.to_vec()
                };

                let count = r.estimate();
                let pct: f64 = if item.is_empty() && !pct_nulls {
                    -1.0
                } else if pct_denom > 0.0 {
                    100.0 * (count as f64) / pct_denom
                } else {
                    0.0
                };
                let pct_str = self.format_percentage(pct, abs_dec_places);

                rank_buf.clear();
                if !item.is_empty() || pct_nulls {
                    let rank_val = (rank_idx as u64) + 1;
                    rank_buf.push_str(itoa_buf.format(rank_val));
                }

                wtr.write_record([
                    header_vec.as_slice(),
                    &display_bytes,
                    itoa_buf.format(count).as_bytes(),
                    pct_str.as_bytes(),
                    rank_buf.as_bytes(),
                ])?;
            }

            if emit_other {
                let total = sketch.total_weight();
                if total > kept_estimate_sum {
                    let other_count = total - kept_estimate_sum;
                    let pct: f64 = if pct_denom > 0.0 {
                        100.0 * (other_count as f64) / pct_denom
                    } else {
                        0.0
                    };
                    let pct_str = self.format_percentage(pct, abs_dec_places);
                    rank_buf.clear();
                    // Match the exact path's Other-row convention: rank is "0"
                    // (sentinel for synthetic rows like Other / *ALL_UNIQUE).
                    // The exact path renders rank 0.0 as the integer "0" via
                    // itoa (see emit_unweighted_csv_rows), so we do the same
                    // here. The label intentionally diverges from exact —
                    // the exact path emits "<other-text> (<unique_count>)" but
                    // FI cannot recover the count of distinct items not in the
                    // top-K (purged items aren't tracked), so we emit the bare
                    // --other-text. This divergence is documented in the
                    // --sketch-method help.
                    rank_buf.push_str(itoa_buf.format(0_u64));
                    wtr.write_record([
                        header_vec.as_slice(),
                        other_label.as_slice(),
                        itoa_buf.format(other_count).as_bytes(),
                        pct_str.as_bytes(),
                        rank_buf.as_bytes(),
                    ])?;
                }
            }
        }

        Ok(wtr.flush()?)
    }
}

/// Helper function to add a field to `field_stats` if it exists
/// Automatically converts any type to appropriate JSON value.
/// NaN and infinity are emitted as JSON null (rather than silently coerced
/// to 0) since JSON cannot represent them and downstream consumers should
/// distinguish "no value" from "zero".
fn add_stat<T: ToString>(field_stats: &mut Vec<FieldStats>, name: &str, value: Option<T>) {
    if let Some(val) = value {
        let val_string = val.to_string();

        // Try to parse as integer first
        let json_value =
            if let Ok(int_val) = atoi_simd::parse::<i64, false, false>(val_string.as_bytes()) {
                JsonValue::Number(int_val.into())
            } else if let Ok(float_val) = fast_float2::parse(&val_string) {
                serde_json::Number::from_f64(float_val).map_or(JsonValue::Null, JsonValue::Number)
            } else {
                // Fall back to string
                JsonValue::String(val_string)
            };

        field_stats.push(FieldStats {
            name:  name.to_string(),
            value: json_value,
        });
    }
}

/// trim leading and trailing whitespace from a byte slice
#[allow(clippy::inline_always)]
#[inline(always)]
pub(crate) fn trim_bs_whitespace(bytes: &[u8]) -> &[u8] {
    let mut start = 0;
    let mut end = bytes.len();

    // safety: use unchecked indexing since we're bounds checking with the while condition
    // Find start by scanning forward
    while start < end {
        let b = unsafe { *bytes.get_unchecked(start) };
        if !b.is_ascii_whitespace() {
            break;
        }
        start += 1;
    }

    // Find end by scanning backward
    while end > start {
        let b = unsafe { *bytes.get_unchecked(end - 1) };
        if !b.is_ascii_whitespace() {
            break;
        }
        end -= 1;
    }

    // safety: This slice is guaranteed to be in bounds due to our index calculations
    unsafe { bytes.get_unchecked(start..end) }
}

/// Evaluate a Luau expression against stats data for a column.
/// Returns `true` if the column should be EXCLUDED from frequency analysis.
/// The Lua VM is passed in to allow reuse across multiple column evaluations.
#[cfg(feature = "luau")]
fn evaluate_stats_filter(
    lua: &mlua::Lua,
    stats_data: &StatsData,
    filter_expression: &str,
) -> Result<bool, String> {
    use mlua::{IntoLua, Value};

    // Set all StatsData fields as Luau globals. mlua's `Table::set` accepts
    // any `IntoLua` value, including `Option<T>` (mapped to Nil for None) and
    // `&str` (mapped to a Lua string).
    //
    // Two macros so String / Option<String> fields can be passed by borrow
    // (`&str` / `Option<&str>`) — `IntoLua for &str` lets Lua copy directly
    // without an intermediate Rust-side `String::clone`. The `bind!` form is
    // for `Copy` field types (numeric, bool, Option<numeric>) where the value
    // is moved out of `*stats_data` for free.
    let globals = lua.globals();
    let set = |name: &str, value: Value| -> Result<(), String> {
        globals
            .set(name, value)
            .map_err(|e| format!("Failed to set {name}: {e}"))
    };
    macro_rules! bind {
        ($name:ident) => {
            set(
                stringify!($name),
                stats_data
                    .$name
                    .into_lua(lua)
                    .map_err(|e| format!("Failed to convert {}: {e}", stringify!($name)))?,
            )?
        };
    }
    macro_rules! bind_str {
        ($name:ident) => {
            set(
                stringify!($name),
                stats_data
                    .$name
                    .as_str()
                    .into_lua(lua)
                    .map_err(|e| format!("Failed to convert {}: {e}", stringify!($name)))?,
            )?
        };
    }
    macro_rules! bind_opt_str {
        ($name:ident) => {
            set(
                stringify!($name),
                stats_data
                    .$name
                    .as_deref()
                    .into_lua(lua)
                    .map_err(|e| format!("Failed to convert {}: {e}", stringify!($name)))?,
            )?
        };
    }

    // Basic fields
    bind_str!(field);
    // 'type' is a reserved keyword in Rust, so we use r#type
    set(
        "type",
        stats_data
            .r#type
            .as_str()
            .into_lua(lua)
            .map_err(|e| format!("Failed to convert type: {e}"))?,
    )?;
    bind!(is_ascii);

    // Counts (non-optional)
    bind!(cardinality);
    bind!(nullcount);

    // Optional numeric / string fields
    bind!(sum);
    bind_opt_str!(min);
    bind_opt_str!(max);
    bind!(range);
    bind_opt_str!(sort_order);

    // String length stats
    bind!(min_length);
    bind!(max_length);
    bind!(sum_length);
    bind!(avg_length);
    bind!(stddev_length);
    bind!(variance_length);
    bind!(cv_length);

    // Numeric stats
    bind!(mean);
    bind!(sem);
    bind!(stddev);
    bind!(variance);
    bind!(cv);

    // Sign counts
    bind!(n_negative);
    bind!(n_zero);
    bind!(n_positive);

    // Precision
    bind!(max_precision);

    // Ratios
    bind!(sparsity);
    bind!(uniqueness_ratio);

    // Distribution stats
    bind!(mad);
    bind!(lower_outer_fence);
    bind!(lower_inner_fence);
    bind!(q1);
    bind!(q2_median);
    bind!(q3);
    bind!(iqr);
    bind!(upper_inner_fence);
    bind!(upper_outer_fence);
    bind!(skewness);

    // Mode / Antimode
    bind_opt_str!(mode);
    bind!(mode_count);
    bind!(mode_occurrences);
    bind_opt_str!(antimode);
    bind!(antimode_count);
    bind!(antimode_occurrences);

    // Wrap the expression in a return statement to get the result
    let wrapped_expr = format!("return {filter_expression}");

    // Evaluate the expression
    let result: Value = lua
        .load(&wrapped_expr)
        .eval()
        .map_err(|e| format!("Failed to evaluate filter expression: {e}"))?;

    // Convert result to boolean
    match result {
        Value::Boolean(b) => Ok(b),
        Value::Nil => Ok(false), // nil is treated as false (don't exclude)
        _ => Err(format!(
            "Filter expression must return a boolean, got: {result:?}"
        )),
    }
}
