static USAGE: &str = r#"
Analyze a SQL query against CSV file caches (stats, moarstats, frequency) to produce a
performance score with actionable optimization suggestions BEFORE running the query.

Accepts the same input/SQL arguments as sqlp. Outputs a human-readable performance report
(default) or JSON (--json). Supports Polars mode (default) and DuckDB mode (--duckdb).

Scoring factors include:
  * Query plan analysis (EXPLAIN output from Polars or DuckDB)
  * Type optimization (column types vs. usage in query)
  * Join key cardinality and data distribution
  * Filter selectivity from frequency cache
  * Query anti-pattern detection (SELECT *, missing LIMIT, cartesian joins, etc.)
  * Infrastructure checks (index files, cache freshness)

Caches are auto-generated when missing:
  * stats cache via `qsv stats --everything --stats-jsonl`
  * frequency cache via `qsv frequency --frequency-jsonl`

Examples:

  # Score a simple filter query against a single CSV file
  $ qsv scoresql data.csv "SELECT * FROM data WHERE col1 > 10"

  # Output the score report as JSON instead of the default human-readable format
  $ qsv scoresql --json data.csv "SELECT col1, col2 FROM data ORDER BY col1"

  # Score a join query across two CSV files
  $ qsv scoresql data.csv data2.csv "SELECT * FROM data JOIN data2 ON data.id = data2.id"

  # Use DuckDB for query plan analysis instead of Polars
  $ qsv scoresql --duckdb data.csv "SELECT * FROM data WHERE status = 'active'"

  # Use _t_N aliases just like sqlp (see sqlp documentation)
  $ qsv scoresql data.csv data2.csv "SELECT * FROM _t_1 JOIN _t_2 ON _t_1.id = _t_2.id"

  # Score a query from a SQL script file (only the last query is scored)
  $ qsv scoresql data.csv script.sql

For more examples, see https://github.com/dathere/qsv/blob/master/tests/test_scoresql.rs.
See also https://github.com/dathere/qsv/wiki/SQL-and-Polars#scoresql

Usage:
    qsv scoresql [options] <input>... <sql>
    qsv scoresql --help

scoresql arguments:
    input                     The CSV file/s to analyze. Use '-' for standard input.
                              If input is a directory, all files in the directory will
                              be read as input.
                              If the input is a file with a '.infile-list' extension,
                              the file will be read as a list of input files.

    sql                       The SQL query to score/analyze.
                              If the query ends with ".sql", it will be read as a
                              SQL script file, with single-line "--" comments stripped.
                              If the script has multiple queries separated by ";",
                              only the last non-empty query is scored.

scoresql options:
    --json                    Output results as JSON instead of human-readable report.
    --duckdb                  Use DuckDB for query plan analysis instead of Polars.
                              Uses the QSV_DUCKDB_PATH environment variable if set,
                              otherwise looks for "duckdb" in PATH.
    --try-parsedates          Automatically try to parse dates/datetimes and time.
    --infer-len <arg>         Number of rows to scan when inferring schema.
                              [default: 10000]
    --ignore-errors           Ignore errors when parsing CSVs.
    --truncate-ragged-lines   Truncate lines with more fields than the header.

Common options:
    -h, --help                Display this message
    -o, --output <file>       Write output to <file> instead of stdout.
    -d, --delimiter <arg>     The field delimiter for reading CSV data.
                              Must be a single character. [default: ,]
    -q, --quiet               Do not print informational messages to stderr.
"#;

use std::{
    borrow::Cow,
    env,
    fmt::Write as FmtWrite,
    io::{Read, Write},
    path::{Path, PathBuf},
    process::Command,
    sync::OnceLock,
};

use foldhash::{HashMap, HashMapExt};
use polars::{prelude::*, sql::SQLContext};
use serde::{Deserialize, Serialize};

use crate::{CliResult, cmd::joinp::tsvssv_delim, config::Delimiter, util, util::process_input};

#[derive(Deserialize, Clone)]
struct Args {
    arg_input:                  Vec<PathBuf>,
    arg_sql:                    String,
    flag_json:                  bool,
    flag_duckdb:                bool,
    flag_try_parsedates:        bool,
    flag_infer_len:             usize,
    flag_ignore_errors:         bool,
    flag_truncate_ragged_lines: bool,
    flag_output:                Option<String>,
    flag_delimiter:             Option<Delimiter>,
    flag_quiet:                 bool,
}

// ── Scoring constants ──────────────────────────────────────────────────────

const MAX_SCORE: u32 = 100;
const WEIGHT_TYPE_OPT: u32 = 20;
const WEIGHT_JOIN_CARD: u32 = 20;
const WEIGHT_FILTER_SEL: u32 = 20;
const WEIGHT_DATA_DIST: u32 = 20;
const WEIGHT_PATTERNS: u32 = 20;

static DUCKDB_PATH: OnceLock<String> = OnceLock::new();

// remove full-line comments starting with "--"
// NOTE: inline trailing comments (e.g., `SELECT 1 -- comment`) are not stripped
static COMMENT_REGEX: std::sync::LazyLock<regex::Regex> =
    std::sync::LazyLock::new(|| regex::Regex::new(r"(?m)^\s*--.*$").unwrap());

// ── Output structs ─────────────────────────────────────────────────────────

#[derive(Serialize)]
struct ScoreReport {
    score:        u32,
    max_score:    u32,
    rating:       String,
    plan:         String,
    breakdown:    Vec<ScoreBreakdown>,
    suggestions:  Vec<String>,
    cache_status: CacheStatus,
}

#[derive(Serialize)]
struct ScoreBreakdown {
    category: String,
    score:    u32,
    max:      u32,
    detail:   String,
}

#[derive(Serialize)]
struct CacheStatus {
    stats_available:     Vec<String>,
    stats_missing:       Vec<String>,
    frequency_available: Vec<String>,
    frequency_missing:   Vec<String>,
    index_available:     Vec<String>,
    index_missing:       Vec<String>,
}

// ── Parsed SQL info ────────────────────────────────────────────────────────

struct SqlInfo {
    has_select_star:  bool,
    has_order_by:     bool,
    has_limit:        bool,
    has_join:         bool,
    has_where:        bool,
    has_subquery:     bool,
    where_columns:    Vec<String>,
    /// `(column, optional_literal)` pairs from `col OP literal` predicates in
    /// the WHERE clause. The literal is `Some` when the right-hand side is a
    /// quoted string, a number, or one of the SQL keyword literals
    /// `TRUE`/`FALSE`/`NULL`; bare identifiers are intentionally excluded so
    /// `a.x = b.y` doesn't masquerade as a value lookup. `None` is reserved
    /// for future predicate types where no literal can be extracted.
    ///
    /// Keyword vs string literals are tagged via [`WhereLiteral`] so that
    /// `WHERE col = NULL` and `WHERE col = 'NULL'` get different lookup
    /// semantics in `score_filter_selectivity` — the former matches empty
    /// frequency-cache cells, the latter only the literal string `"NULL"`.
    where_predicates: Vec<(String, Option<WhereLiteral>)>,
    join_columns:     Vec<String>,
    order_columns:    Vec<String>,
}

/// Distinguishes SQL keyword literals (`TRUE`/`FALSE`/`NULL`) from regular
/// values so `score_filter_selectivity` can normalize keyword forms (case-
/// insensitive match for booleans, empty-string match for nulls) without
/// affecting quoted strings whose payload happens to be the same word.
#[derive(Clone, Debug)]
enum WhereLiteral {
    /// SQL keyword: `TRUE`, `FALSE`, or `NULL`. Stored as upper-case canonical
    /// form so callers can match with `==`.
    Keyword(String),
    /// Quoted string (already un-escaped) or numeric literal. Compared with
    /// exact string equality against frequency-cache values.
    Value(String),
}

// ── Cache data per input file ──────────────────────────────────────────────

struct InputCacheData {
    file_path:      PathBuf,
    table_name:     String,
    stats:          Vec<crate::cmd::stats::StatsData>,
    has_freq_cache: bool,
    has_index:      bool,
    freq_entries:   Vec<FreqEntry>,
}

#[derive(Clone)]
#[allow(dead_code)]
struct FreqEntry {
    field:       String,
    cardinality: u64,
    frequencies: Vec<FreqValue>,
}

#[derive(Clone)]
#[allow(dead_code)]
struct FreqValue {
    value:      String,
    count:      u64,
    percentage: f64,
}

// ════════════════════════════════════════════════════════════════════════════
// run()
// ════════════════════════════════════════════════════════════════════════════

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

    let tmpdir = tempfile::tempdir()?;
    args.arg_input = process_input(args.arg_input, &tmpdir, "")?;

    if args.arg_input.is_empty() {
        return fail_incorrectusage_clierror!("No input files provided.");
    }

    let delim = if let Some(delimiter) = args.flag_delimiter {
        delimiter.as_byte()
    } else if let Ok(delim) = env::var("QSV_DEFAULT_DELIMITER") {
        Delimiter::decode_delimiter(&delim)?.as_byte()
    } else {
        b','
    };

    // ── 1. Resolve table names & aliases ───────────────────────────────────
    let mut table_aliases: HashMap<String, String> = HashMap::with_capacity(args.arg_input.len());
    let mut lossy_table_name = Cow::default();
    let mut table_name;
    let mut table_names: Vec<String> = Vec::with_capacity(args.arg_input.len());
    // O(1) duplicate-stem detection — `table_names.iter().position()` would be
    // O(n²) when scoring a directory expanded to many CSVs by `process_input`.
    // The map stores the first input index where each stem was seen so the
    // error message can still report both positions.
    let mut seen_stems: HashMap<String, usize> = HashMap::with_capacity(args.arg_input.len());

    for (idx, table) in args.arg_input.iter().enumerate() {
        table_name = Path::new(table)
            .file_stem()
            .and_then(std::ffi::OsStr::to_str)
            .unwrap_or_else(|| {
                lossy_table_name = table.to_string_lossy();
                &lossy_table_name
            });
        // Reject duplicate file stems — otherwise the second registration silently
        // overwrites the first in `table_aliases` / `SQLContext::register`, and the
        // user gets a confusing "table not found" error from polars/duckdb later.
        if let Some(&prev_idx) = seen_stems.get(table_name) {
            return fail_incorrectusage_clierror!(
                "Duplicate table name '{table_name}' from inputs #{a} and #{b}. Inputs must have \
                 unique file stems (the part of the file name before the extension).",
                a = prev_idx + 1,
                b = idx + 1,
            );
        }
        seen_stems.insert(table_name.to_string(), idx);
        table_aliases.insert(table_name.to_string(), format!("_t_{}", idx + 1));
        table_names.push(table_name.to_string());
    }

    // ── 2. Parse SQL for patterns ──────────────────────────────────────────
    // check if the input is a SQL script file (ends with .sql)
    if Path::new(&args.arg_sql)
        .extension()
        .is_some_and(|ext| ext.eq_ignore_ascii_case("sql"))
    {
        let mut file = std::fs::File::open(&args.arg_sql)?;
        let mut sql_script = String::new();
        file.read_to_string(&mut sql_script)?;

        let sql_script = COMMENT_REGEX.replace_all(&sql_script, "");

        // split by ";", take the last non-empty query, and trim it
        args.arg_sql = sql_script
            .split(';')
            .map(|s| s.trim().to_string())
            .rfind(|s| !s.is_empty())
            .unwrap_or_default();
    }

    let sql_info = parse_sql(&args.arg_sql);

    // ── 3. Load / generate caches ──────────────────────────────────────────
    let mut input_caches: Vec<InputCacheData> = Vec::with_capacity(args.arg_input.len());

    for (idx, input_path) in args.arg_input.iter().enumerate() {
        let tname = &table_names[idx];
        let cache = load_or_generate_caches(input_path, tname, delim, args.flag_quiet)?;
        input_caches.push(cache);
    }

    // ── 4. Get query plan ──────────────────────────────────────────────────
    let plan = if args.flag_duckdb {
        get_duckdb_plan(&args, &table_names)?
    } else {
        get_polars_plan(&args, &table_aliases, delim)?
    };

    // ── 5. Scoring ─────────────────────────────────────────────────────────
    let mut suggestions: Vec<String> = Vec::new();

    let type_score = score_type_optimization(&sql_info, &input_caches, &mut suggestions);
    let join_score = score_join_cardinality(&sql_info, &input_caches, &mut suggestions);
    let filter_score = score_filter_selectivity(&sql_info, &input_caches, &mut suggestions);
    let dist_score = score_data_distribution(&sql_info, &input_caches, &mut suggestions);
    let pattern_score = score_query_patterns(&sql_info, &plan, &mut suggestions);

    // Infrastructure suggestions
    for cache in &input_caches {
        if !cache.has_index {
            suggestions.push(format!(
                "Run `qsv index {}` to create an index for faster access",
                cache.file_path.display()
            ));
        }
    }

    let total_score = type_score.0 + join_score.0 + filter_score.0 + dist_score.0 + pattern_score.0;

    let rating = match total_score {
        90..=100 => "Excellent",
        75..=89 => "Good",
        50..=74 => "Fair",
        25..=49 => "Poor",
        _ => "Very Poor",
    }
    .to_string();

    // ── 6. Build cache status ──────────────────────────────────────────────
    let cache_status = CacheStatus {
        stats_available:     input_caches
            .iter()
            .filter(|c| !c.stats.is_empty())
            .map(|c| c.table_name.clone())
            .collect(),
        stats_missing:       input_caches
            .iter()
            .filter(|c| c.stats.is_empty())
            .map(|c| c.table_name.clone())
            .collect(),
        frequency_available: input_caches
            .iter()
            .filter(|c| c.has_freq_cache)
            .map(|c| c.table_name.clone())
            .collect(),
        frequency_missing:   input_caches
            .iter()
            .filter(|c| !c.has_freq_cache)
            .map(|c| c.table_name.clone())
            .collect(),
        index_available:     input_caches
            .iter()
            .filter(|c| c.has_index)
            .map(|c| c.table_name.clone())
            .collect(),
        index_missing:       input_caches
            .iter()
            .filter(|c| !c.has_index)
            .map(|c| c.table_name.clone())
            .collect(),
    };

    let breakdown = vec![
        ScoreBreakdown {
            category: "Type optimization".to_string(),
            score:    type_score.0,
            max:      WEIGHT_TYPE_OPT,
            detail:   type_score.1,
        },
        ScoreBreakdown {
            category: "Join cardinality".to_string(),
            score:    join_score.0,
            max:      WEIGHT_JOIN_CARD,
            detail:   join_score.1,
        },
        ScoreBreakdown {
            category: "Filter selectivity".to_string(),
            score:    filter_score.0,
            max:      WEIGHT_FILTER_SEL,
            detail:   filter_score.1,
        },
        ScoreBreakdown {
            category: "Data distribution".to_string(),
            score:    dist_score.0,
            max:      WEIGHT_DATA_DIST,
            detail:   dist_score.1,
        },
        ScoreBreakdown {
            category: "Query patterns".to_string(),
            score:    pattern_score.0,
            max:      WEIGHT_PATTERNS,
            detail:   pattern_score.1,
        },
    ];

    let report = ScoreReport {
        score: total_score,
        max_score: MAX_SCORE,
        rating,
        plan,
        breakdown,
        suggestions,
        cache_status,
    };

    // ── 7. Output ──────────────────────────────────────────────────────────
    let output = if args.flag_json {
        serde_json::to_string_pretty(&report)?
    } else {
        format_human_report(&report)
    };

    if let Some(output_path) = &args.flag_output {
        let mut file = std::fs::File::create(output_path)?;
        file.write_all(output.as_bytes())?;
    } else {
        woutinfo!("{output}");
    }

    Ok(())
}

// ════════════════════════════════════════════════════════════════════════════
// SQL Parsing (lightweight, pattern-based)
// ════════════════════════════════════════════════════════════════════════════

fn parse_sql(sql: &str) -> SqlInfo {
    // Mask string-literal contents with spaces (preserving byte positions) so that
    // keyword/column scanners don't pick up SQL syntax from inside literals like
    // `WHERE col = 'WHERE' OR ...`. Predicate extraction below uses the *raw* SQL
    // because it needs to recover the literal values.
    let masked = mask_string_literals(sql);
    let upper = masked.to_ascii_uppercase();
    let tokens: Vec<&str> = upper.split_whitespace().collect();

    #[allow(clippy::missing_asserts_for_indexing)]
    let has_select_star = tokens.windows(2).any(|w| w[0] == "SELECT" && w[1] == "*");

    #[allow(clippy::missing_asserts_for_indexing)]
    let has_order_by = tokens.windows(2).any(|w| w[0] == "ORDER" && w[1] == "BY");

    let has_limit = tokens.contains(&"LIMIT");

    let has_join = tokens.contains(&"JOIN");
    let has_where = tokens.contains(&"WHERE");

    // Detect subqueries by looking for SELECT inside parentheses.
    // Tracks quote state to avoid false positives from string literals
    // like `WHERE col = '(SELECT ...'`, and checks word boundaries to
    // avoid matching identifiers like `SELECTIVITY`.
    let has_subquery = {
        let mut depth = 0i32;
        let mut in_quote = false;
        let mut found = false;
        let bytes = upper.as_bytes();
        let select_bytes = b"SELECT";
        for (i, &b) in bytes.iter().enumerate() {
            // Toggle quote state; '' (SQL escaped quote) works correctly
            // by toggle symmetry (on then off).
            if b == b'\'' {
                in_quote = !in_quote;
                continue;
            }
            if in_quote {
                continue;
            }
            if b == b'(' {
                depth += 1;
            } else if b == b')' {
                depth -= 1;
            } else if depth > 0
                && b == b'S'
                && (i == 0 || (!bytes[i - 1].is_ascii_alphanumeric() && bytes[i - 1] != b'_'))
                && bytes.get(i..i + select_bytes.len()) == Some(select_bytes)
            {
                // Check word boundary before and after SELECT — exclude '_' too
                // so that identifiers like `_SELECT(` aren't mistaken for a subquery.
                let after = i + select_bytes.len();
                if after >= bytes.len()
                    || bytes[after].is_ascii_whitespace()
                    || bytes[after] == b'('
                {
                    found = true;
                    break;
                }
            }
        }
        found
    };

    // Extract columns/predicates. Pattern-based scanners run on `masked`
    // (string-literal-safe); predicate extraction needs the raw SQL so the
    // literal values inside quotes are preserved.
    let where_columns = extract_columns_after_keyword(&masked, "WHERE");
    let where_predicates = extract_where_predicates(sql);
    let join_columns = extract_join_columns(&masked);
    let order_columns = extract_columns_after_keyword(&masked, "ORDER BY");

    SqlInfo {
        has_select_star,
        has_order_by,
        has_limit,
        has_join,
        has_where,
        has_subquery,
        where_columns,
        where_predicates,
        join_columns,
        order_columns,
    }
}

/// Split a token on SQL comparison operators (=, <, >, <=, >=, <>, !=)
/// so that `status='active'` becomes `["status", "'active'"]` and
/// `a.id=b.id` becomes `["a.id", "b.id"]`.
fn split_on_operators(token: &str) -> Vec<&str> {
    // Split on operator chars, keeping only non-empty parts
    let mut parts = Vec::new();
    let mut start = 0;
    let bytes = token.as_bytes();
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'=' || bytes[i] == b'<' || bytes[i] == b'>' || bytes[i] == b'!' {
            if start < i {
                parts.push(&token[start..i]);
            }
            // Skip the operator (and any following = for <=, >=, !=, <>)
            i += 1;
            if i < bytes.len() && (bytes[i] == b'=' || bytes[i] == b'>') {
                i += 1;
            }
            start = i;
        } else {
            i += 1;
        }
    }
    if start < token.len() {
        parts.push(&token[start..]);
    }
    parts
}

fn extract_columns_after_keyword(sql: &str, keyword: &str) -> Vec<String> {
    let upper = sql.to_ascii_uppercase();
    let Some(pos) = upper.find(keyword) else {
        return Vec::new();
    };
    let after = &sql[pos + keyword.len()..];
    // Take tokens until next SQL keyword
    let stop_keywords = [
        "SELECT",
        "FROM",
        "WHERE",
        "ORDER",
        "GROUP",
        "HAVING",
        "LIMIT",
        "JOIN",
        "ON",
        "UNION",
        "EXCEPT",
        "INTERSECT",
    ];

    let mut columns = Vec::new();
    for token in after.split_whitespace() {
        // First split on comparison operators to handle col=value patterns
        for part in split_on_operators(token) {
            let clean = part.trim_matches(|c: char| !c.is_alphanumeric() && c != '_' && c != '.');
            let upper_clean = clean.to_ascii_uppercase();
            if stop_keywords.contains(&upper_clean.as_str()) {
                return columns;
            }
            if !clean.is_empty()
                && ![
                    "AND", "OR", "NOT", "BY", "ASC", "DESC", "IS", "NULL", "IN", "BETWEEN", "LIKE",
                ]
                .contains(&upper_clean.as_str())
                && !clean.starts_with('\'')
                && !clean.starts_with('"')
                && clean.parse::<f64>().is_err()
            {
                // Strip table prefix (e.g. "data.col1" -> "col1")
                let col = if let Some(dot_pos) = clean.rfind('.') {
                    &clean[dot_pos + 1..]
                } else {
                    clean
                };
                if !col.is_empty() {
                    columns.push(col.to_string());
                }
            }
        }
    }
    columns
}

fn extract_join_columns(sql: &str) -> Vec<String> {
    let upper = sql.to_ascii_uppercase();
    let mut columns = Vec::new();

    // Look for ON clause
    for (i, _) in upper.match_indices(" ON ") {
        let after = &sql[i + 4..];
        'on_clause: for token in after.split_whitespace() {
            // Split on operators to handle a.id=b.id patterns
            for part in split_on_operators(token) {
                let clean =
                    part.trim_matches(|c: char| !c.is_alphanumeric() && c != '_' && c != '.');
                let upper_clean = clean.to_ascii_uppercase();
                if ["WHERE", "ORDER", "GROUP", "HAVING", "LIMIT", "JOIN"]
                    .contains(&upper_clean.as_str())
                {
                    break 'on_clause;
                }
                if clean.contains('.')
                    && let Some(col) = clean.split('.').next_back()
                    && !col.is_empty()
                    && !["AND", "OR"].contains(&upper_clean.as_str())
                {
                    columns.push(col.to_string());
                }
            }
        }
    }

    // Look for USING clause
    for (i, _) in upper.match_indices("USING") {
        let after = &sql[i + 5..];
        // Search for the closing paren only after the opening one to avoid a
        // panic on malformed SQL where ')' appears before '(' in `after`.
        if let Some(paren_start) = after.find('(')
            && let Some(rel_end) = after[paren_start + 1..].find(')')
        {
            let paren_end = paren_start + 1 + rel_end;
            let cols = &after[paren_start + 1..paren_end];
            for col in cols.split(',') {
                let col = col.trim();
                if !col.is_empty() {
                    columns.push(col.to_string());
                }
            }
        }
    }

    // Deduplicate (case-insensitive) to avoid double-counting the same join key
    columns.sort_unstable_by_key(|a| a.to_lowercase());
    columns.dedup_by(|a, b| a.eq_ignore_ascii_case(b));
    columns
}

/// Replace the contents of single-quoted string literals with spaces, preserving
/// byte positions. Used so that keyword-based parsers (`extract_columns_after_keyword`,
/// `extract_join_columns`, etc.) don't pick up SQL syntax from inside literals.
/// SQL-escaped quotes (`''`) are preserved as two spaces. Multi-byte UTF-8 chars
/// inside a literal are replaced with N spaces (N = byte length) so subsequent
/// byte-offset slicing remains valid.
fn mask_string_literals(sql: &str) -> String {
    let mut out = String::with_capacity(sql.len());
    let mut chars = sql.chars().peekable();
    let mut in_quote = false;
    while let Some(c) = chars.next() {
        if c == '\'' {
            if in_quote && chars.peek() == Some(&'\'') {
                // SQL-escaped quote inside a string — mask both
                out.push(' ');
                out.push(' ');
                chars.next();
                continue;
            }
            in_quote = !in_quote;
            out.push('\'');
        } else if in_quote {
            for _ in 0..c.len_utf8() {
                out.push(' ');
            }
        } else {
            out.push(c);
        }
    }
    out
}

#[allow(clippy::items_after_statements)]
fn extract_where_predicates(sql: &str) -> Vec<(String, Option<WhereLiteral>)> {
    // Locate the WHERE clause boundary on the *masked* SQL so a literal
    // `WHERE` inside a string (`SELECT 'WHERE foo' AS x FROM t WHERE col = 'bar'`)
    // doesn't shift the start position. The masked version preserves byte
    // offsets, so positions found on it are valid byte indices into `sql`.
    let masked = mask_string_literals(sql);

    // Word-boundary, case-insensitive WHERE locator. A bare `find("WHERE")` on
    // the upper-cased text would also match the substring inside identifiers
    // like `SOMEWHERE`, misclassifying queries.
    static WHERE_LOC_REGEX: std::sync::LazyLock<regex::Regex> =
        std::sync::LazyLock::new(|| regex::Regex::new(r"(?i)\bWHERE\b").unwrap());
    let Some(m) = WHERE_LOC_REGEX.find(&masked) else {
        return Vec::new();
    };
    let after_start = m.end();
    let masked_after = &masked[after_start..];

    // Word-boundary, whitespace-tolerant stop-keyword matcher. The previous
    // `" ORDER BY "` substring scan failed when clauses were separated by
    // newlines, tabs, or punctuation (common in formatted SQL).
    static STOP_KW_REGEX: std::sync::LazyLock<regex::Regex> = std::sync::LazyLock::new(|| {
        regex::Regex::new(
            r"(?i)\b(?:ORDER\s+BY|GROUP\s+BY|HAVING|LIMIT|UNION|EXCEPT|INTERSECT|WINDOW)\b",
        )
        .unwrap()
    });
    let end = STOP_KW_REGEX
        .find(masked_after)
        .map_or(masked_after.len(), |m| m.start());

    // Use raw `sql` for the actual scan so quoted literal values survive.
    let where_clause = &sql[after_start..after_start + end];

    static WHERE_PATTERN: std::sync::LazyLock<regex::Regex> = std::sync::LazyLock::new(|| {
        // Either a string literal (consumed but ignored — group 1 is None),
        // or a predicate `col OP literal`. The RHS is restricted to literal
        // forms only; bare identifiers like `b.y` are intentionally not
        // matched so we never confuse a column reference with a literal value.
        // Operator alternation lists multi-char operators first.
        //
        // NOTE: only standard SQL `''` quote-escaping is recognized — same
        // assumption `mask_string_literals` makes. PostgreSQL `E'...'` strings
        // and backslash escapes (`\'`) inside literals are NOT supported; if
        // future dialects need them, update both this regex and the masker.
        regex::Regex::new(
                r"'(?:[^']|'')*'|(\w+(?:\.\w+)?)\s*(?:<=|>=|<>|!=|=|<|>)\s*('(?:[^']|'')*'|-?\d+(?:\.\d+)?|(?i:TRUE|FALSE|NULL))",
            )
            .unwrap()
    });

    let mut preds = Vec::new();
    for cap in WHERE_PATTERN.captures_iter(where_clause) {
        // String-literal-only matches have no group 1 — they're swallowed by
        // the alternation so we advance past them without false-extracting
        // predicates from inside.
        let Some(col_match) = cap.get(1) else {
            continue;
        };
        let val_raw = cap.get(2).unwrap().as_str();
        let col_full = col_match.as_str();

        // Skip purely numeric LHS (e.g. `1 = 1` tautologies)
        if col_full.parse::<f64>().is_ok() {
            continue;
        }

        // Strip table prefix (`a.id` -> `id`)
        let col = col_full.rsplit('.').next().unwrap_or(col_full).to_string();
        let upper_col = col.to_ascii_uppercase();

        // Skip SQL keywords that the regex might pick up as identifiers
        if [
            "AND", "OR", "NOT", "IS", "NULL", "IN", "BETWEEN", "LIKE", "TRUE", "FALSE",
        ]
        .contains(&upper_col.as_str())
        {
            continue;
        }

        // Tag the literal so score_filter_selectivity can apply keyword
        // normalization (TRUE/FALSE/NULL) only to the *unquoted* form.
        // `'TRUE'` / `'FALSE'` / `'NULL'` (the quoted strings) are stored as
        // `Value` and compared with exact string equality.
        let value = if val_raw.starts_with('\'') && val_raw.ends_with('\'') && val_raw.len() >= 2 {
            // Quoted string — un-escape '' and store as Value
            WhereLiteral::Value(val_raw[1..val_raw.len() - 1].replace("''", "'"))
        } else {
            // Unquoted: number or TRUE/FALSE/NULL keyword
            let upper = val_raw.to_ascii_uppercase();
            if matches!(upper.as_str(), "TRUE" | "FALSE" | "NULL") {
                WhereLiteral::Keyword(upper)
            } else {
                WhereLiteral::Value(val_raw.to_string())
            }
        };
        preds.push((col, Some(value)));
    }
    preds
}

// ════════════════════════════════════════════════════════════════════════════
// Cache Loading / Generation
// ════════════════════════════════════════════════════════════════════════════

fn load_or_generate_caches(
    input_path: &Path,
    table_name: &str,
    delim: u8,
    quiet: bool,
) -> CliResult<InputCacheData> {
    let canonical = input_path.canonicalize()?;

    // Check for stats cache
    let stats_path = canonical.with_extension("stats.csv.data.jsonl");
    let stats = if stats_path.exists() && is_cache_fresh(&stats_path, &canonical) {
        load_stats_cache(&stats_path)?
    } else {
        if !quiet {
            winfo!("Generating stats cache for {table_name}...");
        }
        generate_stats_cache(&canonical, delim)?;
        if stats_path.exists() {
            load_stats_cache(&stats_path)?
        } else {
            Vec::new()
        }
    };

    // Check for frequency cache
    let freq_path = canonical.with_extension("freq.csv.data.jsonl");
    let (has_freq_cache, freq_entries) =
        if freq_path.exists() && is_cache_fresh(&freq_path, &canonical) {
            (true, load_freq_cache(&freq_path)?)
        } else {
            if !quiet {
                winfo!("Generating frequency cache for {table_name}...");
            }
            generate_freq_cache(&canonical, delim)?;
            if freq_path.exists() {
                (true, load_freq_cache(&freq_path)?)
            } else {
                (false, Vec::new())
            }
        };

    // Check for index
    let idx_path = canonical.with_extension("csv.idx");
    let has_index = idx_path.exists();

    Ok(InputCacheData {
        file_path: input_path.to_path_buf(),
        table_name: table_name.to_string(),
        stats,
        has_freq_cache,
        has_index,
        freq_entries,
    })
}

fn is_cache_fresh(cache_path: &Path, input_path: &Path) -> bool {
    let Ok(cache_meta) = cache_path.metadata() else {
        return false;
    };
    let Ok(input_meta) = input_path.metadata() else {
        return false;
    };
    let Ok(cache_mtime) = cache_meta.modified() else {
        return false;
    };
    let Ok(input_mtime) = input_meta.modified() else {
        return false;
    };
    cache_mtime >= input_mtime
}

fn load_stats_cache(stats_path: &Path) -> CliResult<Vec<crate::cmd::stats::StatsData>> {
    let content = std::fs::read_to_string(stats_path)?;
    let mut stats = Vec::new();
    // stats --stats-jsonl has no metadata header — every line is a StatsData record
    for line in content.lines() {
        if line.is_empty() {
            continue;
        }
        if let Ok(sd) = serde_json::from_str::<crate::cmd::stats::StatsData>(line) {
            stats.push(sd);
        }
    }
    Ok(stats)
}

fn load_freq_cache(freq_path: &Path) -> CliResult<Vec<FreqEntry>> {
    let content = std::fs::read_to_string(freq_path)?;
    let mut entries = Vec::new();
    // The first non-empty line is metadata — skip it.
    // Tracking via a flag (rather than `i == 0`) handles a leading blank line
    // robustly, so the metadata is never misparsed as data.
    let mut metadata_seen = false;
    for line in content.lines() {
        if line.is_empty() {
            continue;
        }
        if !metadata_seen {
            metadata_seen = true;
            continue;
        }
        // Deserialize using serde_json into a generic value
        if let Ok(val) = serde_json::from_str::<serde_json::Value>(line) {
            let field = val["field"].as_str().unwrap_or("").to_string();
            let cardinality = val["cardinality"].as_u64().unwrap_or(0);
            let frequencies = if let Some(freqs) = val["frequencies"].as_array() {
                freqs
                    .iter()
                    .map(|f| FreqValue {
                        value:      f["value"].as_str().unwrap_or("").to_string(),
                        count:      f["count"].as_u64().unwrap_or(0),
                        percentage: f["percentage"].as_f64().unwrap_or(0.0),
                    })
                    .collect()
            } else {
                Vec::new()
            };
            entries.push(FreqEntry {
                field,
                cardinality,
                frequencies,
            });
        }
    }
    Ok(entries)
}

fn generate_stats_cache(input_path: &Path, delim: u8) -> CliResult<()> {
    let qsv_bin = env::current_exe()?;
    let mut cmd = Command::new(&qsv_bin);
    cmd.args(["stats", "--everything", "--stats-jsonl"]);
    if delim != b',' {
        // Safety: qsv's --delimiter only accepts single-byte ASCII delimiters
        debug_assert!(delim.is_ascii(), "delimiter must be ASCII");
        cmd.args(["--delimiter", &(delim as char).to_string()]);
    }
    let output = cmd.arg(input_path).output()?;
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        log::warn!("Stats cache generation failed: {stderr}");
        wwarn!(
            "Stats cache generation failed for {p} - scoring may be inaccurate: {stderr}",
            p = input_path.display()
        );
    }
    Ok(())
}

fn generate_freq_cache(input_path: &Path, delim: u8) -> CliResult<()> {
    let qsv_bin = env::current_exe()?;
    let mut cmd = Command::new(&qsv_bin);
    cmd.args(["frequency", "--frequency-jsonl"]);
    if delim != b',' {
        // Safety: qsv's --delimiter only accepts single-byte ASCII delimiters
        debug_assert!(delim.is_ascii(), "delimiter must be ASCII");
        cmd.args(["--delimiter", &(delim as char).to_string()]);
    }
    let output = cmd.arg(input_path).output()?;
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        log::warn!("Frequency cache generation failed: {stderr}");
        wwarn!(
            "Frequency cache generation failed for {p} - scoring may be inaccurate: {stderr}",
            p = input_path.display()
        );
    }
    Ok(())
}

// ════════════════════════════════════════════════════════════════════════════
// Query Plan Generation
// ════════════════════════════════════════════════════════════════════════════

fn get_polars_plan(
    args: &Args,
    table_aliases: &HashMap<String, String>,
    delim: u8,
) -> CliResult<String> {
    let mut ctx = SQLContext::new();

    let optflags = OptFlags::PROJECTION_PUSHDOWN
        | OptFlags::PREDICATE_PUSHDOWN
        | OptFlags::CLUSTER_WITH_COLUMNS
        | OptFlags::TYPE_COERCION
        | OptFlags::SIMPLIFY_EXPR
        | OptFlags::SLICE_PUSHDOWN
        | OptFlags::COMM_SUBPLAN_ELIM
        | OptFlags::COMM_SUBEXPR_ELIM
        | OptFlags::ROW_ESTIMATE
        | OptFlags::FAST_PROJECTION;

    let mut lossy_table_name = Cow::default();
    let mut table_name_str;

    for table in &args.arg_input {
        table_name_str = Path::new(table)
            .file_stem()
            .and_then(std::ffi::OsStr::to_str)
            .unwrap_or_else(|| {
                lossy_table_name = table.to_string_lossy();
                &lossy_table_name
            });

        let separator = tsvssv_delim(table, delim);
        let table_plpath = PlRefPath::new(&*table.to_string_lossy());

        let lf = LazyCsvReader::new(table_plpath)
            .with_has_header(true)
            .with_missing_is_null(true)
            .with_separator(separator)
            .with_infer_schema_length(Some(args.flag_infer_len))
            .with_try_parse_dates(args.flag_try_parsedates)
            .with_ignore_errors(args.flag_ignore_errors)
            .with_truncate_ragged_lines(args.flag_truncate_ragged_lines)
            .finish()?;

        ctx.register(table_name_str, lf.with_optimizations(optflags));
    }

    // Replace _t_N aliases with quoted table names using a single combined regex.
    // Alternation is longest-first so _t_10 is matched before _t_1.
    // The pattern also matches single-quoted string literals (with SQL-escaped quotes)
    // so we can leave them untouched — otherwise an alias appearing inside a string
    // literal (e.g. `SELECT '_t_1' AS label FROM _t_1`) would be silently corrupted.
    let mut alias_pairs: Vec<_> = table_aliases.iter().collect();
    alias_pairs.sort_by_key(|(_, alias)| std::cmp::Reverse(alias.len()));

    let query = if alias_pairs.is_empty() {
        args.arg_sql.clone()
    } else {
        let alternatives: String = alias_pairs
            .iter()
            .map(|(_, alias)| regex::escape(alias))
            .collect::<Vec<_>>()
            .join("|");
        let pattern = format!(r"'(?:[^'\\]|\\'|''|\\\\)*'|\b(?:{alternatives})\b");
        let re = regex::Regex::new(&pattern)
            .map_err(|e| format!("Failed to compile alias regex: {e}"))?;

        let alias_lookup: HashMap<&str, &str> = alias_pairs
            .iter()
            .map(|(tname, talias)| (talias.as_str(), tname.as_str()))
            .collect();

        re.replace_all(&args.arg_sql, |caps: &regex::Captures| {
            let matched = caps.get(0).unwrap().as_str();
            // Leave string literals untouched
            if matched.starts_with('\'') {
                matched.to_string()
            } else {
                format!(r#""{}""#, alias_lookup[matched])
            }
        })
        .into_owned()
    };

    let explain_query = format!("EXPLAIN {query}");
    match ctx.execute(&explain_query) {
        Ok(lf) => match lf.collect() {
            Ok(df) => Ok(format!("{df}")),
            Err(e) => {
                fail_clierror!(
                    "Failed to generate query plan.\n\nSQL: {sql}\n\nError: {e}\n\nHint: Check \
                     your SQL syntax.",
                    sql = args.arg_sql
                )
            },
        },
        Err(e) => {
            fail_clierror!(
                "Failed to execute SQL query.\n\nSQL: {sql}\n\nError: {e}\n\nHint: Check your SQL \
                 syntax.",
                sql = args.arg_sql
            )
        },
    }
}

fn get_duckdb_plan(args: &Args, table_names: &[String]) -> CliResult<String> {
    let duckdb_path = get_duckdb_path()?;

    // Translate _t_N aliases and table names to read_csv('path').
    // Process replacements longest-first to avoid partial matches
    // (e.g., "data" matching inside "data2").
    let mut replacements: Vec<(String, String)> = Vec::new();
    for (idx, input) in args.arg_input.iter().enumerate() {
        let alias = format!("_t_{}", idx + 1);
        let table_name = &table_names[idx];
        let escaped = input.to_string_lossy().replace('\'', "''");
        let read_csv = format!("read_csv_auto('{escaped}')");
        replacements.push((alias, read_csv.clone()));
        // Only add table_name replacement if it differs from the alias
        if *table_name != format!("_t_{}", idx + 1) {
            replacements.push((table_name.clone(), read_csv));
        }
    }
    // Sort by length descending so longer names are replaced first
    replacements.sort_by_key(|(name, _)| std::cmp::Reverse(name.len()));
    // Deduplicate in case a table name equals an alias
    replacements.dedup_by(|a, b| a.0 == b.0);

    // Build a single combined regex for one-pass replacement.
    // The regex matches single-quoted strings first (to skip them) OR table names/aliases.
    // This prevents replacing table names that appear inside SQL string literals
    // (e.g., inside an existing read_csv_auto('...') call in a .sql script).
    let query = if replacements.is_empty() {
        args.arg_sql.clone()
    } else {
        let alternatives: String = replacements
            .iter()
            .map(|(name, _)| regex::escape(name))
            .collect::<Vec<_>>()
            .join("|");
        // Match single-quoted strings (including '' and \' escaped quotes) OR table names
        let pattern = format!(r"'(?:[^'\\]|\\'|''|\\\\)*'|\b(?:{alternatives})\b");
        let re = regex::Regex::new(&pattern)
            .map_err(|e| format!("Failed to compile DuckDB alias regex: {e}"))?;

        let replacement_lookup: HashMap<&str, &str> = replacements
            .iter()
            .map(|(name, target)| (name.as_str(), target.as_str()))
            .collect();

        re.replace_all(&args.arg_sql, |caps: &regex::Captures| {
            let matched = caps.get(0).unwrap().as_str();
            // If the match starts with a quote, it's a string literal — keep it as-is
            if matched.starts_with('\'') {
                matched.to_string()
            } else {
                replacement_lookup[matched].to_string()
            }
        })
        .into_owned()
    };

    let explain_query = format!("EXPLAIN {query}");
    let output = Command::new(duckdb_path)
        .arg("-csv")
        .arg("-c")
        .arg(&explain_query)
        .output()
        .map_err(|e| format!("Error executing DuckDB: {e}"))?;

    if output.status.success() {
        Ok(String::from_utf8_lossy(&output.stdout).to_string())
    } else {
        let stderr = String::from_utf8_lossy(&output.stderr);
        fail_clierror!(
            "Failed to execute SQL query with DuckDB.\n\nSQL: {sql}\n\nError: {stderr}\n\nHint: \
             Check your SQL syntax.",
            sql = args.arg_sql
        )
    }
}

fn get_duckdb_path() -> CliResult<String> {
    if let Some(path) = DUCKDB_PATH.get() {
        return Ok(path.clone());
    }

    let duckdb_path = if let Ok(explicit_path) = env::var("QSV_DUCKDB_PATH") {
        // Env var is set — validate the explicit path
        let path = Path::new(&explicit_path);
        if !path.exists() {
            return fail_clierror!("DuckDB binary not found at path: {explicit_path}");
        }
        if !path.is_file() {
            return fail_clierror!("DuckDB path is not a file: {explicit_path}");
        }
        if !util::is_executable(&explicit_path)? {
            return fail_clierror!("DuckDB path is not executable: {explicit_path}");
        }
        explicit_path
    } else {
        // Env var not set — try to find "duckdb" in PATH using a cross-platform approach.
        // On Unix we use "which", on Windows we use "where".
        let which_cmd = cfg_select! {
            target_os = "windows" => "where",
            _ => "which",
        };

        if let Some(resolved) = Command::new(which_cmd)
            .arg("duckdb")
            .output()
            .ok()
            .filter(|o| o.status.success())
            .and_then(|o| {
                let s = String::from_utf8(o.stdout).ok()?;
                // `where` on Windows may return multiple lines; take the first
                let trimmed = s.lines().next().unwrap_or("").trim().to_string();
                if trimmed.is_empty() {
                    None
                } else {
                    Some(trimmed)
                }
            })
        {
            resolved
        } else {
            return fail_clierror!(
                "DuckDB not found. Either set QSV_DUCKDB_PATH to the DuckDB binary path or ensure \
                 \"duckdb\" is in your PATH."
            );
        }
    };

    let _ = DUCKDB_PATH.set(duckdb_path.clone());
    Ok(duckdb_path)
}

// ════════════════════════════════════════════════════════════════════════════
// Scoring Functions — each returns (score, detail_string)
// ════════════════════════════════════════════════════════════════════════════

fn score_type_optimization(
    sql_info: &SqlInfo,
    caches: &[InputCacheData],
    suggestions: &mut Vec<String>,
) -> (u32, String) {
    if caches.iter().all(|c| c.stats.is_empty()) {
        return (
            WEIGHT_TYPE_OPT,
            "No stats available — full score assumed".to_string(),
        );
    }

    let mut score = WEIGHT_TYPE_OPT;
    let mut issues = Vec::new();

    // Check columns used in WHERE/JOIN — ideally they should be numeric types for comparisons
    let comparison_cols: Vec<&str> = sql_info
        .where_columns
        .iter()
        .chain(sql_info.join_columns.iter())
        .map(String::as_str)
        .collect();

    for cache in caches {
        for stat in &cache.stats {
            let col_upper = stat.field.to_ascii_uppercase();
            let is_referenced = comparison_cols
                .iter()
                .any(|c| c.to_ascii_uppercase() == col_upper);

            if is_referenced && stat.r#type == "String" {
                // String type used in comparison — potential issue
                score = score.saturating_sub(3);
                issues.push(format!(
                    "{}.{} used in comparison but typed as String",
                    cache.table_name, stat.field
                ));
                suggestions.push(format!(
                    "Column '{}.{}' is typed as String but used in comparisons — consider casting \
                     or ensuring proper type inference",
                    cache.table_name, stat.field
                ));
            }
        }
    }

    let detail = if issues.is_empty() {
        "all referenced columns have appropriate types".to_string()
    } else {
        issues.join("; ")
    };

    (score, detail)
}

fn score_join_cardinality(
    sql_info: &SqlInfo,
    caches: &[InputCacheData],
    suggestions: &mut Vec<String>,
) -> (u32, String) {
    if !sql_info.has_join {
        return (WEIGHT_JOIN_CARD, "no joins in query".to_string());
    }

    if sql_info.join_columns.is_empty() {
        // Potential cartesian join!
        suggestions.push(
            "Possible cartesian join detected — missing ON/USING clause. This will produce a \
             cross product of all rows."
                .to_string(),
        );
        return (
            0,
            "possible cartesian join — missing join condition".to_string(),
        );
    }

    let mut score = WEIGHT_JOIN_CARD;
    let mut details = Vec::new();

    for join_col in &sql_info.join_columns {
        let col_upper = join_col.to_ascii_uppercase();
        for cache in caches {
            if let Some(stat) = cache
                .stats
                .iter()
                .find(|s| s.field.to_ascii_uppercase() == col_upper)
            {
                // Check cardinality
                if stat.cardinality < 10 {
                    score = score.saturating_sub(3);
                    details.push(format!(
                        "{}.{} low cardinality ({}) — may cause skew",
                        cache.table_name, stat.field, stat.cardinality
                    ));
                    suggestions.push(format!(
                        "Join key '{}.{}' has very low cardinality ({}) — may cause data skew in \
                         join results",
                        cache.table_name, stat.field, stat.cardinality
                    ));
                }

                // Check nulls in join key
                if stat.nullcount > 0 {
                    let sparsity = stat.sparsity.unwrap_or(0.0);
                    if sparsity > 0.1 {
                        score = score.saturating_sub(2);
                        details.push(format!(
                            "{}.{} has {:.0}% nulls — wasteful in joins",
                            cache.table_name,
                            stat.field,
                            sparsity * 100.0
                        ));
                        suggestions.push(format!(
                            "Join key '{}.{}' has {:.0}% null values — consider filtering nulls \
                             before joining",
                            cache.table_name,
                            stat.field,
                            sparsity * 100.0
                        ));
                    }
                }
            }
        }
    }

    let detail = if details.is_empty() {
        "good cardinality on join keys".to_string()
    } else {
        details.join("; ")
    };

    (score, detail)
}

fn score_filter_selectivity(
    sql_info: &SqlInfo,
    caches: &[InputCacheData],
    suggestions: &mut Vec<String>,
) -> (u32, String) {
    if !sql_info.has_where {
        return (
            WEIGHT_FILTER_SEL / 2,
            "no WHERE clause — full table scan".to_string(),
        );
    }

    let mut score = WEIGHT_FILTER_SEL;
    let mut details = Vec::new();

    // For each WHERE column, prefer to look up the *actual filter value* in the
    // frequency cache. If we can match it, we know the selectivity. If the
    // filter has no extractable literal (e.g. `col IS NULL`, complex
    // expressions), fall back to a soft "high-skew" note without penalizing —
    // the old "top value > 70%" heuristic over-penalized selective filters
    // that happened to target rare values in skewed columns.
    //
    // Dedup key uses borrowed (cache_idx, &str) so we don't clone two Strings
    // per (predicate, cache) iteration in this hot scoring path. The owned
    // `String` for the upper-case column is built once outside the cache loop.
    let mut handled: foldhash::HashSet<(usize, &str)> = foldhash::HashSet::default();
    let upper_cols: Vec<String> = sql_info
        .where_predicates
        .iter()
        .map(|(c, _)| c.to_ascii_uppercase())
        .collect();

    for ((_, val_opt), col_upper) in sql_info.where_predicates.iter().zip(upper_cols.iter()) {
        for (cache_idx, cache) in caches.iter().enumerate() {
            if !handled.insert((cache_idx, col_upper.as_str())) {
                continue;
            }
            if let Some(freq) = cache
                .freq_entries
                .iter()
                .find(|f| f.field.eq_ignore_ascii_case(col_upper))
            {
                // Look up the literal in the frequency cache. SQL keyword
                // literals get normalization, but only when they came from
                // the *unquoted* form (`WHERE col = NULL`) — quoted strings
                // (`WHERE col = 'NULL'`) keep exact-equality semantics.
                //   - `NULL`  -> qsv frequency writes nulls as the empty string, so match
                //     `fv.value.is_empty()`
                //   - `TRUE`/`FALSE` -> match case-insensitively against the stringified CSV cell
                //     ("True", "true", ...)
                let matched_pct = val_opt
                    .as_ref()
                    .and_then(|lit| match lit {
                        WhereLiteral::Keyword(k) if k == "NULL" => {
                            freq.frequencies.iter().find(|fv| fv.value.is_empty())
                        },
                        WhereLiteral::Keyword(k) => freq
                            .frequencies
                            .iter()
                            .find(|fv| fv.value.eq_ignore_ascii_case(k)),
                        WhereLiteral::Value(v) => freq.frequencies.iter().find(|fv| fv.value == *v),
                    })
                    .map(|fv| fv.percentage);

                if let Some(pct) = matched_pct {
                    if pct > 70.0 {
                        score = score.saturating_sub(5);
                        details.push(format!(
                            "{}.{} filter value matches {:.0}% of rows (low selectivity)",
                            cache.table_name, freq.field, pct
                        ));
                        suggestions.push(format!(
                            "Column '{}.{}' filter matches ~{:.0}% of rows — consider a more \
                             selective predicate",
                            cache.table_name, freq.field, pct
                        ));
                    }
                    // Otherwise the filter is selective: no penalty.
                } else if let Some(top) = freq.frequencies.first()
                    && top.percentage > 70.0
                    && top.value != "<ALL_UNIQUE>"
                    && top.value != "<HIGH_CARDINALITY>"
                {
                    // Could not match the filter value (e.g. literal not in the top-N
                    // sample, or compared against another column). Note the skew, but
                    // do not penalize — the filter may well be selecting a rare value.
                    details.push(format!(
                        "{}.{} is highly skewed (top value '{}' = {:.0}%) — selectivity depends \
                         on filter value",
                        cache.table_name, freq.field, top.value, top.percentage
                    ));
                }
            }
        }
    }

    // Columns referenced in WHERE without an extractable predicate value
    // (e.g. `col IS NULL`, function calls): only emit a soft note for high-skew
    // columns; never penalize.
    let upper_where_cols: Vec<String> = sql_info
        .where_columns
        .iter()
        .map(|c| c.to_ascii_uppercase())
        .collect();
    for col_upper in &upper_where_cols {
        for (cache_idx, cache) in caches.iter().enumerate() {
            if handled.contains(&(cache_idx, col_upper.as_str())) {
                continue;
            }
            if let Some(freq) = cache
                .freq_entries
                .iter()
                .find(|f| f.field.eq_ignore_ascii_case(col_upper))
                && let Some(top) = freq.frequencies.first()
                && top.percentage > 70.0
                && top.value != "<ALL_UNIQUE>"
                && top.value != "<HIGH_CARDINALITY>"
            {
                details.push(format!(
                    "{}.{} is highly skewed (top value '{}' = {:.0}%) — selectivity depends on \
                     filter value",
                    cache.table_name, freq.field, top.value, top.percentage
                ));
            }
        }
    }

    let detail = if details.is_empty() {
        "filter columns have reasonable selectivity".to_string()
    } else {
        details.join("; ")
    };

    (score, detail)
}

fn score_data_distribution(
    sql_info: &SqlInfo,
    caches: &[InputCacheData],
    suggestions: &mut Vec<String>,
) -> (u32, String) {
    let relevant_cols: Vec<&str> = sql_info
        .join_columns
        .iter()
        .chain(sql_info.where_columns.iter())
        .chain(sql_info.order_columns.iter())
        .map(String::as_str)
        .collect();

    if relevant_cols.is_empty() {
        return (
            WEIGHT_DATA_DIST,
            "no columns to check distribution for".to_string(),
        );
    }

    let mut score = WEIGHT_DATA_DIST;
    let mut details = Vec::new();

    for cache in caches {
        for stat in &cache.stats {
            let col_upper = stat.field.to_ascii_uppercase();
            let is_referenced = relevant_cols
                .iter()
                .any(|c| c.to_ascii_uppercase() == col_upper);
            if !is_referenced {
                continue;
            }

            // Check skewness
            if let Some(skewness) = stat.skewness
                && skewness.abs() > 2.0
            {
                score = score.saturating_sub(2);
                details.push(format!(
                    "{}.{} highly skewed (skewness={:.2})",
                    cache.table_name, stat.field, skewness
                ));
            }

            // Check gini coefficient (from moarstats)
            if let Some(gini) = stat.gini_coefficient
                && gini > 0.5
            {
                score = score.saturating_sub(2);
                details.push(format!(
                    "{}.{} high inequality (gini={:.2})",
                    cache.table_name, stat.field, gini
                ));
                suggestions.push(format!(
                    "Column '{}.{}' has high inequality (gini={:.2}) — may cause performance \
                     issues in joins/aggregations",
                    cache.table_name, stat.field, gini
                ));
            }

            // Check entropy (from moarstats)
            if let Some(entropy) = stat.normalized_entropy
                && entropy < 0.3
            {
                score = score.saturating_sub(1);
                details.push(format!(
                    "{}.{} low entropy ({:.2}) — data is highly concentrated",
                    cache.table_name, stat.field, entropy
                ));
            }
        }
    }

    let detail = if details.is_empty() {
        "data distribution looks balanced".to_string()
    } else {
        details.join("; ")
    };

    (score, detail)
}

fn score_query_patterns(
    sql_info: &SqlInfo,
    plan: &str,
    suggestions: &mut Vec<String>,
) -> (u32, String) {
    let mut score = WEIGHT_PATTERNS;
    let mut issues = Vec::new();

    // SELECT * anti-pattern
    if sql_info.has_select_star {
        score = score.saturating_sub(4);
        issues.push("SELECT * detected".to_string());
        suggestions
            .push("Replace SELECT * with specific columns to reduce memory usage".to_string());
    }

    // ORDER BY without LIMIT on what could be large results
    if sql_info.has_order_by && !sql_info.has_limit {
        score = score.saturating_sub(3);
        issues.push("ORDER BY without LIMIT".to_string());
        suggestions.push(
            "Add LIMIT to ORDER BY queries to avoid sorting the entire result set".to_string(),
        );
    }

    // Nested subqueries (could be CTEs)
    if sql_info.has_subquery {
        score = score.saturating_sub(2);
        issues.push("nested subquery detected".to_string());
        suggestions.push(
            "Consider using CTEs (WITH clause) instead of nested subqueries for better \
             readability and potential optimization"
                .to_string(),
        );
    }

    // Cartesian join (JOIN without ON)
    if sql_info.has_join && sql_info.join_columns.is_empty() {
        score = score.saturating_sub(5);
        issues.push("cartesian join — missing ON/WHERE clause".to_string());
    }

    // Check plan for full table scans
    let plan_upper = plan.to_ascii_uppercase();
    if plan_upper.contains("FULL SCAN") || plan_upper.contains("TABLE SCAN") {
        score = score.saturating_sub(2);
        issues.push("full table scan in query plan".to_string());
    }

    let detail = if issues.is_empty() {
        "no anti-patterns detected".to_string()
    } else {
        issues.join("; ")
    };

    (score, detail)
}

// ════════════════════════════════════════════════════════════════════════════
// Output Formatting
// ════════════════════════════════════════════════════════════════════════════

fn format_human_report(report: &ScoreReport) -> String {
    let mut out = String::with_capacity(2048);

    let _ = write!(
        out,
        "Score: {}/{} ({})\n\n",
        report.score, report.max_score, report.rating
    );

    out.push_str("=== Query Plan ===\n");
    out.push_str(&report.plan);
    out.push('\n');
    if !report.plan.ends_with('\n') {
        out.push('\n');
    }

    out.push_str("\n=== Scoring Breakdown ===\n");
    for b in &report.breakdown {
        let _ = writeln!(
            out,
            "  {:25} {:>2}/{:<2}  - {}",
            format!("{}:", b.category),
            b.score,
            b.max,
            b.detail
        );
    }

    if !report.suggestions.is_empty() {
        out.push_str("\n=== Suggestions ===\n");
        for (i, s) in report.suggestions.iter().enumerate() {
            let _ = writeln!(out, "  {}. {s}", i + 1);
        }
    }

    // Cache status summary
    if !report.cache_status.stats_missing.is_empty()
        || !report.cache_status.frequency_missing.is_empty()
        || !report.cache_status.index_missing.is_empty()
    {
        out.push_str("\n=== Cache Status ===\n");
        if !report.cache_status.stats_missing.is_empty() {
            let _ = writeln!(
                out,
                "  Stats missing: {}",
                report.cache_status.stats_missing.join(", ")
            );
        }
        if !report.cache_status.frequency_missing.is_empty() {
            let _ = writeln!(
                out,
                "  Frequency missing: {}",
                report.cache_status.frequency_missing.join(", ")
            );
        }
        if !report.cache_status.index_missing.is_empty() {
            let _ = writeln!(
                out,
                "  Index missing: {}",
                report.cache_status.index_missing.join(", ")
            );
        }
    }

    out
}
