perf-sentinel-core 0.17.0

//! Ingestion for `MySQL` Performance Schema statement digests.
//!
//! Parses CSV or JSON exports of `performance_schema.events_statements_summary_by_digest`
//! into a `MySqlStatReport` with top-N rankings by total execution time, call
//! count, mean execution time, and rows examined.
//!
//! Timer columns (`SUM_TIMER_WAIT`, `AVG_TIMER_WAIT`) arrive in picoseconds and
//! are converted to milliseconds at parse time. Like `pg_stat_statements`, the
//! digest view has no `trace_id`, it provides a complementary database-level
//! view of SQL hotspots.

use crate::detect::Finding;
use crate::normalize::sql::normalize_sql;
use serde::{Deserialize, Serialize};

/// Picoseconds per millisecond: `MySQL` timer columns are picoseconds.
const PS_PER_MS: f64 = 1e9;

/// A single entry from `events_statements_summary_by_digest`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct MySqlStatEntry {
    /// Original digest text (already `?`-parameterized by `MySQL`).
    pub query: String,
    /// Template after `perf-sentinel` SQL normalization.
    pub normalized_template: String,
    /// Schema the digest was observed in (`SCHEMA_NAME`), when present.
    pub schema_name: Option<String>,
    /// Number of times the statement was executed (`COUNT_STAR`).
    pub calls: u64,
    /// Total execution time in milliseconds (from `SUM_TIMER_WAIT` picoseconds).
    pub total_exec_time_ms: f64,
    /// Mean execution time in milliseconds (from `AVG_TIMER_WAIT` picoseconds).
    pub mean_exec_time_ms: f64,
    /// Total rows sent to clients (`SUM_ROWS_SENT`).
    pub rows_sent: u64,
    /// Total rows examined by the storage engine (`SUM_ROWS_EXAMINED`).
    pub rows_examined: u64,
    /// Whether this template was also seen in trace-based findings.
    #[serde(default)]
    pub seen_in_traces: bool,
}

/// A ranking of digest entries by a specific criterion.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct MySqlStatRanking {
    /// Label describing the ranking criterion (e.g., "top by `total_exec_time`").
    pub label: String,
    /// Entries sorted by the criterion, limited to `top_n`.
    pub entries: Vec<MySqlStatEntry>,
}

/// Report produced from Performance Schema digest analysis.
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct MySqlStatReport {
    /// Total number of entries parsed.
    pub total_entries: usize,
    /// Number of top entries per ranking.
    pub top_n: usize,
    /// Rankings in a stable order: by `total_exec_time`, by `calls`,
    /// by `mean_exec_time`, by `rows_examined`. Consumers that index by
    /// position (e.g., the HTML dashboard's `mysql_stat` sub-switcher)
    /// rely on this ordering not changing. New rankings are appended,
    /// existing indices are never reassigned.
    pub rankings: Vec<MySqlStatRanking>,
    /// Matched share from the trace cross-reference. `None` when no
    /// trace set was provided. Additive, absent from the JSON when
    /// unset.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub trace_match: Option<crate::ingest::pg_stat::TraceMatchSummary>,
}

/// Detected input format.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MySqlStatFormat {
    Csv,
    Json,
}

/// Errors that can occur during Performance Schema digest parsing.
///
/// `#[non_exhaustive]` for SemVer-minor variant additions.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum MySqlStatError {
    #[error("payload too large: {size} bytes exceeds maximum of {max} bytes")]
    PayloadTooLarge { size: usize, max: usize },
    #[error("CSV parse error at line {line}: {detail}")]
    CsvParse { line: usize, detail: String },
    #[error("JSON parse error: {0}")]
    JsonParse(String),
    #[error("missing required column: {0}")]
    MissingColumn(String),
    #[error("empty input")]
    EmptyInput,
    #[error("Prometheus request failed: {0}")]
    PrometheusRequest(String),
    #[error("Prometheus response format error: {0}")]
    PrometheusFormat(String),
}

/// Raw JSON entry matching common digest export shapes: raw
/// `performance_schema` UPPERCASE column names or their lowercase twins.
/// Timer fields stay in picoseconds here, converted on mapping.
#[derive(Deserialize)]
struct RawJsonEntry {
    // Option: performance_schema keeps a catch-all aggregation row with
    // DIGEST_TEXT = NULL once the digest table saturates; that row is
    // skipped instead of failing the whole export.
    #[serde(default, alias = "DIGEST_TEXT")]
    digest_text: Option<String>,
    #[serde(default, alias = "SCHEMA_NAME")]
    schema_name: Option<String>,
    #[serde(alias = "COUNT_STAR")]
    count_star: u64,
    #[serde(alias = "SUM_TIMER_WAIT")]
    sum_timer_wait: f64,
    #[serde(alias = "AVG_TIMER_WAIT")]
    avg_timer_wait: f64,
    #[serde(default, alias = "SUM_ROWS_SENT")]
    sum_rows_sent: u64,
    #[serde(default, alias = "SUM_ROWS_EXAMINED")]
    sum_rows_examined: u64,
}

/// Detect whether the input is CSV or JSON.
///
/// Peeks at the first non-whitespace byte: `[` or `{` indicates JSON,
/// otherwise CSV. Returns `Csv` as fallback for empty input; the caller
/// should validate non-emptiness separately.
#[must_use]
pub fn detect_mysql_stat_format(raw: &[u8]) -> MySqlStatFormat {
    let trimmed = raw.iter().position(|&b| !b.is_ascii_whitespace());
    match trimmed.map(|i| raw[i]) {
        Some(b'[' | b'{') => MySqlStatFormat::Json,
        _ => MySqlStatFormat::Csv,
    }
}

/// Parse Performance Schema digest data from raw bytes.
///
/// Auto-detects CSV vs JSON format. Normalizes each digest through the
/// SQL normalizer for consistency with trace-based analysis (backticked
/// identifiers survive, `IN (?, ?, ...)` collapses to `IN (?)`).
///
/// # Errors
///
/// Returns an error if the payload exceeds `max_size`, the input is empty,
/// or parsing fails.
pub fn parse_mysql_stat(
    raw: &[u8],
    max_size: usize,
) -> Result<Vec<MySqlStatEntry>, MySqlStatError> {
    if raw.len() > max_size {
        return Err(MySqlStatError::PayloadTooLarge {
            size: raw.len(),
            max: max_size,
        });
    }
    if raw.is_empty() || raw.iter().all(|&b| b.is_ascii_whitespace()) {
        return Err(MySqlStatError::EmptyInput);
    }

    let text = std::str::from_utf8(raw).map_err(|e| MySqlStatError::CsvParse {
        line: 0,
        detail: format!("invalid UTF-8: {e}"),
    })?;

    match detect_mysql_stat_format(raw) {
        MySqlStatFormat::Csv => parse_csv(text),
        MySqlStatFormat::Json => parse_json(text),
    }
}

/// Generate rankings from parsed entries.
///
/// Produces four rankings in a stable order: by total execution time,
/// by call count, by mean execution time, by rows examined. Each ranking
/// contains at most `top_n` entries. Index-based sorting, same clone
/// trade-off as `rank_pg_stat` (one-shot path, never per-event).
///
/// Downstream consumers rely on the rankings appearing at the documented
/// positions, new rankings are always appended and existing indices
/// never reassign.
#[must_use]
pub fn rank_mysql_stat(entries: &[MySqlStatEntry], top_n: usize) -> MySqlStatReport {
    let total_entries = entries.len();

    let top_n_by = |cmp: fn(&MySqlStatEntry, &MySqlStatEntry) -> std::cmp::Ordering,
                    label: &str|
     -> MySqlStatRanking {
        let mut indices: Vec<usize> = (0..entries.len()).collect();
        indices.sort_by(|&a, &b| cmp(&entries[a], &entries[b]));
        indices.truncate(top_n);
        MySqlStatRanking {
            label: label.to_string(),
            entries: indices.iter().map(|&i| entries[i].clone()).collect(),
        }
    };

    let by_total_time = top_n_by(
        |a, b| {
            b.total_exec_time_ms
                .partial_cmp(&a.total_exec_time_ms)
                .unwrap_or(std::cmp::Ordering::Equal)
        },
        "top by total_exec_time",
    );

    let by_calls = top_n_by(|a, b| b.calls.cmp(&a.calls), "top by calls");

    let by_mean_time = top_n_by(
        |a, b| {
            b.mean_exec_time_ms
                .partial_cmp(&a.mean_exec_time_ms)
                .unwrap_or(std::cmp::Ordering::Equal)
        },
        "top by mean_exec_time",
    );

    // Rows examined is MySQL's I/O-cost signal: a high examined-to-sent
    // ratio flags full scans and missing indexes, the closest analog to
    // pg's shared-buffer traffic ranking.
    let by_rows_examined = top_n_by(
        |a, b| b.rows_examined.cmp(&a.rows_examined),
        "top by rows_examined",
    );

    MySqlStatReport {
        total_entries,
        top_n,
        rankings: vec![by_total_time, by_calls, by_mean_time, by_rows_examined],
        trace_match: None,
    }
}

/// Cross-reference digest entries with trace-based findings.
///
/// Marks entries whose `normalized_template` matches any finding's
/// pattern template. Both sides are canonicalized first: `MySQL`
/// `DIGEST_TEXT` spaces every token (`` `c` . `name` ``), uppercases
/// keywords and forces backtick quoting, none of which appears in a
/// template normalized from raw application SQL, so an exact string
/// compare would silently never match.
pub fn cross_reference(entries: &mut [MySqlStatEntry], findings: &[Finding]) {
    mark_matching(
        entries,
        findings.iter().map(|f| f.pattern.template.as_str()),
    );
}

/// Cross-reference entries against every normalized SQL template observed
/// in the traces, whether or not a detector fired on it (see
/// [`crate::pipeline::trace_sql_template_counts`]). Same canonicalization
/// as [`cross_reference`], which stays as the findings-only fallback.
pub fn cross_reference_templates<S: std::hash::BuildHasher>(
    entries: &mut [MySqlStatEntry],
    trace_counts: &std::collections::HashMap<String, u64, S>,
) {
    mark_matching(entries, trace_counts.keys().map(String::as_str));
}

fn mark_matching<'a>(entries: &mut [MySqlStatEntry], templates: impl Iterator<Item = &'a str>) {
    let keys: std::collections::HashSet<String> = templates.map(comparison_key).collect();

    for entry in entries {
        if keys.contains(&comparison_key(&entry.normalized_template)) {
            entry.seen_in_traces = true;
        }
    }
}

/// Tally the matched share over the entries' `seen_in_traces` flags,
/// weighted by `calls`. Same caveat as the `pg_stat` variant: digest
/// counters are cumulative since the last reset, this is a matched
/// share, not a sampling ratio.
#[must_use]
pub fn trace_match_summary(
    entries: &[MySqlStatEntry],
) -> crate::ingest::pg_stat::TraceMatchSummary {
    crate::ingest::pg_stat::tally_matches(entries.iter().map(|e| (e.seen_in_traces, e.calls)))
}

/// Punctuation and operator tokens `MySQL` digest text surrounds with
/// spaces while compact application SQL does not.
fn is_token_punct(c: char) -> bool {
    matches!(
        c,
        '.' | ',' | '(' | ')' | '=' | '<' | '>' | ';' | '!' | '+' | '-' | '*' | '/' | '%' | '?'
    )
}

/// Best-effort canonical form for digest-vs-trace template comparison:
/// strip backtick quoting, drop whitespace around punctuation and
/// operators, collapse remaining whitespace runs, lowercase (`MySQL`
/// uppercases keywords in digest text while application SQL usually
/// does not).
///
/// Known ceiling: lowercasing also folds identifiers, so on a
/// case-sensitive server (`lower_case_table_names=0`) two tables that
/// differ only by case share a key and the `[seen in traces]` marker
/// can over-match. Accepted for an informational marker; a
/// keyword-only fold would need a full keyword table.
fn comparison_key(template: &str) -> String {
    let mut out = String::with_capacity(template.len());
    let mut pending_space = false;
    for c in template.chars() {
        if c == '`' {
            continue;
        }
        if c.is_whitespace() {
            pending_space = true;
            continue;
        }
        if pending_space
            && !is_token_punct(c)
            && !out.is_empty()
            // Drop the pending space when the previous emitted char was
            // punctuation too ("a . b" and "a. b" both become "a.b").
            && !out.chars().next_back().is_some_and(is_token_punct)
        {
            out.push(' ');
        }
        pending_space = false;
        out.extend(c.to_lowercase());
    }
    out
}

// ---------------------------------------------------------------------------
// CSV parsing (RFC 4180 subset, row parser shared with pg_stat)
// ---------------------------------------------------------------------------

const MAX_CSV_ROWS: usize = 1_000_000;

use super::pg_stat::parse_csv_row;

/// `SCHEMA_NAME` normalization: `MySQL` renders absent schemas as SQL
/// `NULL` (client exports) or `\N` (`INTO OUTFILE` style dumps).
fn parse_schema_name(value: Option<&String>) -> Option<String> {
    value
        .map(|s| s.trim())
        .filter(|s| !is_null_marker(s))
        .map(ToString::to_string)
}

/// The textual NULL renderings `MySQL` tooling emits for absent values.
fn is_null_marker(s: &str) -> bool {
    s.is_empty() || s.eq_ignore_ascii_case("null") || s == "\\N"
}

fn parse_csv(text: &str) -> Result<Vec<MySqlStatEntry>, MySqlStatError> {
    let mut lines = text.lines();

    let header_line = lines.next().ok_or(MySqlStatError::EmptyInput)?;
    let headers = parse_csv_row(header_line);
    let col = |name: &str| -> Result<usize, MySqlStatError> {
        headers
            .iter()
            .position(|h| h.eq_ignore_ascii_case(name))
            .ok_or_else(|| MySqlStatError::MissingColumn(name.to_string()))
    };

    let digest_idx = col("digest_text")?;
    let calls_idx = col("count_star")?;
    let total_time_idx = col("sum_timer_wait")?;
    let mean_time_idx = col("avg_timer_wait")?;
    let schema_idx = col("schema_name").ok();
    let rows_sent_idx = col("sum_rows_sent").ok();
    let rows_examined_idx = col("sum_rows_examined").ok();

    // Estimate row count from byte length (~100 bytes per row), capped at 100k entries
    let estimated = (text.len() / 100).min(100_000);
    let mut entries = Vec::with_capacity(estimated);
    for (line_num, line) in lines.enumerate() {
        if entries.len() >= MAX_CSV_ROWS {
            return Err(MySqlStatError::CsvParse {
                line: line_num + 2,
                detail: format!("CSV exceeds maximum of {MAX_CSV_ROWS} rows"),
            });
        }
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let fields = parse_csv_row(line);
        let line_num = line_num + 2; // 1-indexed, header is line 1

        let query = fields.get(digest_idx).cloned().unwrap_or_default();
        // Skip the catch-all aggregation row (DIGEST_TEXT is NULL once
        // the digest table saturates) instead of ranking a "NULL" query.
        if is_null_marker(query.trim()) {
            continue;
        }
        let calls = parse_u64(&fields, calls_idx, line_num, "count_star")?;
        // Timer columns are picoseconds and can exceed u64 on aggregated
        // servers, so they parse as f64 from the start.
        let sum_timer_wait = parse_f64(&fields, total_time_idx, line_num, "sum_timer_wait")?;
        let avg_timer_wait = parse_f64(&fields, mean_time_idx, line_num, "avg_timer_wait")?;
        let schema_name = parse_schema_name(schema_idx.and_then(|i| fields.get(i)));
        let rows_sent =
            rows_sent_idx.map_or(Ok(0), |i| parse_u64(&fields, i, line_num, "sum_rows_sent"))?;
        let rows_examined = rows_examined_idx.map_or(Ok(0), |i| {
            parse_u64(&fields, i, line_num, "sum_rows_examined")
        })?;

        let normalized = normalize_sql(&query);

        entries.push(MySqlStatEntry {
            query,
            normalized_template: normalized.template,
            schema_name,
            calls,
            total_exec_time_ms: sum_timer_wait / PS_PER_MS,
            mean_exec_time_ms: avg_timer_wait / PS_PER_MS,
            rows_sent,
            rows_examined,
            seen_in_traces: false,
        });
    }

    if entries.is_empty() {
        return Err(MySqlStatError::EmptyInput);
    }
    Ok(entries)
}

fn parse_u64(
    fields: &[String],
    idx: usize,
    line: usize,
    col_name: &str,
) -> Result<u64, MySqlStatError> {
    let val = fields.get(idx).map_or("", String::as_str).trim();
    val.parse::<u64>().map_err(|_| MySqlStatError::CsvParse {
        line,
        detail: format!("cannot parse '{val}' as integer for column {col_name}"),
    })
}

fn parse_f64(
    fields: &[String],
    idx: usize,
    line: usize,
    col_name: &str,
) -> Result<f64, MySqlStatError> {
    let val = fields.get(idx).map_or("", String::as_str).trim();
    val.parse::<f64>().map_err(|_| MySqlStatError::CsvParse {
        line,
        detail: format!("cannot parse '{val}' as float for column {col_name}"),
    })
}

// ---------------------------------------------------------------------------
// JSON parsing
// ---------------------------------------------------------------------------

fn parse_json(text: &str) -> Result<Vec<MySqlStatEntry>, MySqlStatError> {
    let raw_entries: Vec<RawJsonEntry> =
        serde_json::from_str(text).map_err(|e| MySqlStatError::JsonParse(e.to_string()))?;

    if raw_entries.is_empty() {
        return Err(MySqlStatError::EmptyInput);
    }
    if raw_entries.len() > MAX_CSV_ROWS {
        return Err(MySqlStatError::JsonParse(format!(
            "JSON array exceeds maximum of {MAX_CSV_ROWS} entries (got {})",
            raw_entries.len()
        )));
    }

    let entries: Vec<MySqlStatEntry> = raw_entries
        .into_iter()
        .filter_map(|raw| {
            // Skip the catch-all aggregation row (DIGEST_TEXT NULL).
            let digest_text = raw.digest_text.filter(|d| !is_null_marker(d.trim()))?;
            let normalized = normalize_sql(&digest_text);
            let schema_name = parse_schema_name(raw.schema_name.as_ref());
            Some(MySqlStatEntry {
                query: digest_text,
                normalized_template: normalized.template,
                schema_name,
                calls: raw.count_star,
                total_exec_time_ms: raw.sum_timer_wait / PS_PER_MS,
                mean_exec_time_ms: raw.avg_timer_wait / PS_PER_MS,
                rows_sent: raw.sum_rows_sent,
                rows_examined: raw.sum_rows_examined,
                seen_in_traces: false,
            })
        })
        .collect();

    // Every row was dropped: distinguish "wrong export shape" from a
    // legitimate report instead of returning a silent empty success
    // (rows missing DIGEST_TEXT entirely look exactly like the NULL
    // catch-all row to the filter above).
    if entries.is_empty() {
        return Err(MySqlStatError::JsonParse(
            "no row carried a usable DIGEST_TEXT (wrong column names in the \
             export, or every digest is NULL)"
                .to_string(),
        ));
    }

    Ok(entries)
}

// ── Prometheus scrape path ─────────────────────────────────────────

/// Unit of the ranked time series.
///
/// Performance Schema counts `SUM_TIMER_WAIT` in picoseconds. The
/// `mysqld_exporter` collector converts to seconds, a recording rule or a
/// hand-written exporter usually forwards the column untouched, and reading
/// one for the other is off by a factor of 10^12.
#[cfg(any(feature = "daemon", feature = "tempo"))]
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum MySqlStatTimeUnit {
    /// `mysql_perf_schema_events_statements_seconds_total`, the collector built-in.
    #[default]
    Seconds,
    /// A millisecond column, as an exporter dividing `SUM_TIMER_WAIT` by 10^9 emits.
    Milliseconds,
    /// `SUM_TIMER_WAIT` forwarded as is.
    Picoseconds,
}

#[cfg(any(feature = "daemon", feature = "tempo"))]
impl MySqlStatTimeUnit {
    /// Convert a sample to milliseconds, the unit every entry carries.
    fn to_ms(self, value: f64) -> f64 {
        match self {
            Self::Seconds => value * 1000.0,
            Self::Milliseconds => value,
            Self::Picoseconds => value / 1e9,
        }
    }
}

/// Which series and label carry the Performance Schema digests on a given
/// Prometheus.
///
/// The defaults match `mysqld_exporter`'s `perf_schema.eventsstatements`
/// collector, which is not enabled by default: it needs
/// `--collect.perf_schema.eventsstatements` on the exporter. A deployment
/// scraping through a recording rule or a hand-written exporter names its
/// own series, so neither name is ours to assume.
///
/// `#[non_exhaustive]` so a future knob stays a minor bump rather than a
/// breaking change: external crates build it with [`Self::default`] or
/// [`Self::with_overrides`], never a struct literal.
#[cfg(any(feature = "daemon", feature = "tempo"))]
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct PrometheusMySqlStat {
    /// Series ranked by `topk`, holding the cumulated execution time in seconds.
    pub series: String,
    /// Label carrying the statement digest text. Falls back to `digest` when
    /// absent, which leaves the report with opaque hashes rather than SQL.
    pub query_label: String,
    /// Series holding `COUNT_STAR`. The exporter publishes it as a series of
    /// its own, never as a label, so it takes a second query joined on
    /// `digest`. `None` skips that query and leaves the counts at zero, which
    /// empties the ranking by calls and collapses the mean onto the total.
    pub calls_series: Option<String>,
    /// Series holding `SUM_ROWS_SENT`, fetched and joined like `calls_series`.
    /// `None` leaves `rows_sent` at zero.
    pub rows_sent_series: Option<String>,
    /// Series holding `SUM_ROWS_EXAMINED`, fetched and joined like
    /// `calls_series`. `None` leaves `rows_examined` at zero, which empties the
    /// ranking by rows examined.
    pub rows_examined_series: Option<String>,
    /// Label carrying the schema name. Part of the identity every query folds
    /// on, so naming it wrong merges two schemas into one row with their
    /// execution times summed.
    pub schema_label: String,
    /// Unit of `series`. The collector built-in publishes seconds.
    pub unit: MySqlStatTimeUnit,
}

#[cfg(any(feature = "daemon", feature = "tempo"))]
impl Default for PrometheusMySqlStat {
    fn default() -> Self {
        Self {
            series: "mysql_perf_schema_events_statements_seconds_total".to_string(),
            calls_series: Some("mysql_perf_schema_events_statements_total".to_string()),
            rows_sent_series: Some(
                "mysql_perf_schema_events_statements_rows_sent_total".to_string(),
            ),
            rows_examined_series: Some(
                "mysql_perf_schema_events_statements_rows_examined_total".to_string(),
            ),
            query_label: "digest_text".to_string(),
            schema_label: "schema".to_string(),
            unit: MySqlStatTimeUnit::Seconds,
        }
    }
}

#[cfg(any(feature = "daemon", feature = "tempo"))]
impl PrometheusMySqlStat {
    /// Apply the overrides an operator supplied, keeping the defaults for the
    /// rest. Lives here so the CLI entry points never restate the defaults.
    #[must_use]
    pub fn with_overrides(series: Option<String>, query_label: Option<String>) -> Self {
        let d = Self::default();
        Self {
            series: series.unwrap_or(d.series),
            query_label: query_label.unwrap_or(d.query_label),
            calls_series: d.calls_series,
            rows_sent_series: d.rows_sent_series,
            rows_examined_series: d.rows_examined_series,
            schema_label: d.schema_label,
            unit: d.unit,
        }
    }

    /// Identity every query folds on, and every counter joins on. `digest`
    /// leads because it is the identifier: a row without it is dropped rather
    /// than matched on the schema alone. `instance` and `job` keep two servers
    /// apart, and read as empty when the exporter omits them.
    fn identity(&self) -> [&str; 4] {
        ["digest", self.schema_label.as_str(), "instance", "job"]
    }
}

/// Fetch Performance Schema digests from a Prometheus endpoint.
///
/// The exporter publishes `COUNT_STAR`, `SUM_ROWS_SENT` and
/// `SUM_ROWS_EXAMINED` as series of their own rather than labels, so one query
/// each fetches them and joins on the digest identity. A series set to `None`
/// skips its query and leaves its column at `0` rather than inventing one,
/// which for `calls` also collapses the mean onto the total. A file export
/// stays the one input that needs no exporter collector enabled.
///
/// # Errors
///
/// Returns [`MySqlStatError::PrometheusRequest`] on transport errors, an
/// invalid endpoint, an invalid series name or an invalid auth header, and
/// [`MySqlStatError::PrometheusFormat`] if the response cannot be parsed.
#[cfg(any(feature = "daemon", feature = "tempo"))]
pub async fn fetch_from_prometheus(
    endpoint: &str,
    top_n: usize,
    auth_header: Option<&str>,
    opts: &PrometheusMySqlStat,
) -> Result<Vec<MySqlStatEntry>, MySqlStatError> {
    use crate::ingest::prometheus_scrape as scrape;

    scrape::validate_endpoint(endpoint).map_err(MySqlStatError::PrometheusRequest)?;
    scrape::validate_series_name(&opts.series).map_err(MySqlStatError::PrometheusRequest)?;
    // The counter series land in the same query string, unencoded, so they need
    // the same guard: an operator typo carrying `&` or `#` would smuggle a
    // second parameter into the URL rather than fail.
    for series in [
        opts.calls_series.as_deref(),
        opts.rows_sent_series.as_deref(),
        opts.rows_examined_series.as_deref(),
    ]
    .into_iter()
    .flatten()
    {
        scrape::validate_series_name(series).map_err(MySqlStatError::PrometheusRequest)?;
    }

    // Both labels now land in the `sum by (...)` clause, not only in the
    // response they are read back from.
    for label in [&opts.query_label, &opts.schema_label] {
        scrape::validate_label_name(label).map_err(MySqlStatError::PrometheusRequest)?;
    }

    // The ranked query folds on the same identity every counter joins on, plus
    // the label carrying the statement text.
    let identity = opts.identity();
    let mut group_by = identity.to_vec();
    group_by.push(opts.query_label.as_str());
    let query = scrape::build_topk_query(top_n, &opts.series, &group_by);
    let body =
        scrape::fetch_instant_query(endpoint, &query, auth_header, "perf-sentinel/mysql-stat")
            .await
            .map_err(MySqlStatError::PrometheusRequest)?;

    // One query per counter the digest view carries as a series of its own,
    // issued together since none of them feeds another. A failure is not fatal:
    // the timings are already in hand, and a report ranked by time beats no
    // report at all.
    let (calls, rows_sent, rows_examined) = tokio::join!(
        fetch_counter(
            endpoint,
            auth_header,
            &query,
            opts.calls_series.as_deref(),
            &identity,
        ),
        fetch_counter(
            endpoint,
            auth_header,
            &query,
            opts.rows_sent_series.as_deref(),
            &identity,
        ),
        fetch_counter(
            endpoint,
            auth_header,
            &query,
            opts.rows_examined_series.as_deref(),
            &identity,
        ),
    );
    let counters = ScrapedCounters {
        calls,
        rows_sent,
        rows_examined,
    };

    parse_prometheus_response(&body, &counters, opts)
}

/// The counter responses joined onto the ranked rows, each `None` when its
/// series was not configured or the query failed.
#[cfg(any(feature = "daemon", feature = "tempo"))]
#[derive(Default)]
struct ScrapedCounters {
    calls: Option<bytes::Bytes>,
    rows_sent: Option<bytes::Bytes>,
    rows_examined: Option<bytes::Bytes>,
}

/// Fetch one counter series, intersected with the ranked statements. `None`
/// on an unconfigured series or a failed query, which leaves its column at
/// zero rather than failing a report the timings already carry.
#[cfg(any(feature = "daemon", feature = "tempo"))]
async fn fetch_counter(
    endpoint: &str,
    auth_header: Option<&str>,
    ranked_query: &str,
    series: Option<&str>,
    identity: &[&str],
) -> Option<bytes::Bytes> {
    use crate::ingest::prometheus_scrape as scrape;

    let series = series?;
    let query = scrape::build_counter_query(series, identity, ranked_query);
    match scrape::fetch_instant_query(endpoint, &query, auth_header, "perf-sentinel/mysql-stat")
        .await
    {
        Ok(raw) => Some(raw),
        Err(e) => {
            tracing::warn!(
                series,
                error = %e,
                "counter series unavailable, its column stays at zero"
            );
            None
        }
    }
}

/// Parse a Prometheus instant query response into `MySqlStatEntry` structs.
#[cfg(any(feature = "daemon", feature = "tempo"))]
fn parse_prometheus_response(
    body: &[u8],
    counters: &ScrapedCounters,
    opts: &PrometheusMySqlStat,
) -> Result<Vec<MySqlStatEntry>, MySqlStatError> {
    use crate::ingest::prometheus_scrape as scrape;

    let results = scrape::instant_query_results(body).map_err(MySqlStatError::PrometheusFormat)?;

    // identity -> counter, empty when that query was skipped or failed. Not
    // indexed at all when the ranking came back empty: a database with no
    // digests yet would otherwise warn once per counter about series that are
    // fine.
    let identity = opts.identity();
    let index = |raw: Option<&bytes::Bytes>, series: Option<&String>| match (raw, series) {
        (Some(raw), Some(series)) if !results.is_empty() => {
            scrape::counter_by_labels(raw, &identity, series)
                .map_err(MySqlStatError::PrometheusFormat)
        }
        _ => Ok(std::collections::HashMap::new()),
    };
    let call_counts = index(counters.calls.as_ref(), opts.calls_series.as_ref())?;
    let rows_sent = index(counters.rows_sent.as_ref(), opts.rows_sent_series.as_ref())?;
    let rows_examined = index(
        counters.rows_examined.as_ref(),
        opts.rows_examined_series.as_ref(),
    )?;

    let mut entries = Vec::with_capacity(results.len());
    for result in &results {
        let metric = result.get("metric").unwrap_or(&serde_json::Value::Null);
        // `digest` is the fallback for the same reason `queryid` is on the
        // PostgreSQL side: an opaque identifier still groups, where "unknown"
        // would collapse every row into one.
        let query_text = metric
            .get(opts.query_label.as_str())
            .or_else(|| metric.get("digest"))
            .and_then(|v| v.as_str())
            .unwrap_or("unknown")
            .to_string();

        let schema_name = metric
            .get(opts.schema_label.as_str())
            .and_then(|v| v.as_str())
            .map(str::to_string);

        let total_exec_time_ms = opts.unit.to_ms(scrape::sample_value(result));
        let normalized = normalize_sql(&query_text);
        // One key for every counter: all four queries folded on the same
        // identity, so a row either matches in all of them or in none.
        let key = scrape::identity_key(metric, &identity);
        let counter = |index: &std::collections::HashMap<String, u64>| {
            key.as_ref()
                .and_then(|k| index.get(k).copied())
                .unwrap_or(0)
        };
        let calls = counter(&call_counts);
        #[allow(clippy::cast_precision_loss)]
        let mean_exec_time_ms = if calls > 0 {
            total_exec_time_ms / (calls as f64)
        } else {
            total_exec_time_ms
        };

        entries.push(MySqlStatEntry {
            query: query_text,
            normalized_template: normalized.template,
            schema_name,
            calls,
            total_exec_time_ms,
            mean_exec_time_ms,
            rows_sent: counter(&rows_sent),
            rows_examined: counter(&rows_examined),
            seen_in_traces: false,
        });
    }

    Ok(entries)
}

#[cfg(all(test, any(feature = "daemon", feature = "tempo")))]
mod prometheus_tests {
    use super::{
        MySqlStatTimeUnit, PrometheusMySqlStat, ScrapedCounters, parse_prometheus_response,
    };

    /// Only the call counter came back, the shape every test but the row-join
    /// one exercises.
    fn calls_only(raw: &'static [u8]) -> ScrapedCounters {
        ScrapedCounters {
            calls: Some(bytes::Bytes::from_static(raw)),
            ..Default::default()
        }
    }

    #[test]
    fn the_time_unit_scales_the_sample() {
        // SUM_TIMER_WAIT forwarded untouched is picoseconds: read as seconds
        // it would report 4.5e12 ms for a 4.5 ms statement.
        let timings = br#"{"data":{"result":[
            {"metric":{"digest_text":"SELECT 1","digest":"abc"},"value":[1,"4500000000"]}]}}"#;
        let mut opts = PrometheusMySqlStat {
            unit: MySqlStatTimeUnit::Picoseconds,
            ..Default::default()
        };
        let entries =
            parse_prometheus_response(timings, &ScrapedCounters::default(), &opts).expect("parse");
        assert!((entries[0].total_exec_time_ms - 4.5).abs() < 0.001);

        opts.unit = MySqlStatTimeUnit::Milliseconds;
        let entries =
            parse_prometheus_response(timings, &ScrapedCounters::default(), &opts).expect("parse");
        assert!((entries[0].total_exec_time_ms - 4_500_000_000.0).abs() < 0.001);
    }

    #[test]
    fn the_row_counters_join_like_the_call_counter() {
        // The collector publishes rows sent and rows examined as series of
        // their own too, so the ranking by rows examined has real data rather
        // than an all-zero column.
        let timings = br#"{"data":{"result":[
            {"metric":{"digest_text":"SELECT 1","digest":"abc","schema":"shop"},"value":[1,"4.5"]}]}}"#;
        let counters = ScrapedCounters {
            calls: Some(bytes::Bytes::from_static(
                br#"{"data":{"result":[{"metric":{"digest":"abc","schema":"shop"},"value":[1,"9"]}]}}"#,
            )),
            rows_sent: Some(bytes::Bytes::from_static(
                br#"{"data":{"result":[{"metric":{"digest":"abc","schema":"shop"},"value":[1,"90"]}]}}"#,
            )),
            rows_examined: Some(bytes::Bytes::from_static(
                br#"{"data":{"result":[{"metric":{"digest":"abc","schema":"shop"},"value":[1,"7000"]}]}}"#,
            )),
        };
        let entries =
            parse_prometheus_response(timings, &counters, &PrometheusMySqlStat::default())
                .expect("parse");
        assert_eq!(entries[0].calls, 9);
        assert_eq!(entries[0].rows_sent, 90);
        assert_eq!(entries[0].rows_examined, 7000);
    }

    #[test]
    fn two_servers_stay_two_rows() {
        // One Prometheus scraping several database servers: folding them
        // together would sum times across machines into a row naming none.
        let timings = br#"{"data":{"result":[
            {"metric":{"digest_text":"SELECT 1","digest":"abc","instance":"db1:9104"},"value":[1,"4.5"]},
            {"metric":{"digest_text":"SELECT 1","digest":"abc","instance":"db2:9104"},"value":[1,"1.5"]}]}}"#;
        let counters = ScrapedCounters {
            calls: Some(bytes::Bytes::from_static(
                br#"{"data":{"result":[
                    {"metric":{"digest":"abc","instance":"db1:9104"},"value":[1,"9"]},
                    {"metric":{"digest":"abc","instance":"db2:9104"},"value":[1,"3"]}]}}"#,
            )),
            ..Default::default()
        };
        let entries =
            parse_prometheus_response(timings, &counters, &PrometheusMySqlStat::default())
                .expect("parse");
        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].calls, 9, "each server keeps its own count");
        assert_eq!(entries[1].calls, 3);
    }

    #[test]
    fn a_renamed_schema_label_keeps_the_rows_apart() {
        // A recording rule emitting `db` instead of `schema` would otherwise
        // fold two schemas into one row, with their execution times summed.
        let timings = br#"{"data":{"result":[
            {"metric":{"digest_text":"SELECT 1","digest":"abc","db":"shop"},"value":[1,"4.5"]},
            {"metric":{"digest_text":"SELECT 1","digest":"abc","db":"crm"},"value":[1,"1.5"]}]}}"#;
        let counters = ScrapedCounters {
            calls: Some(bytes::Bytes::from_static(
                br#"{"data":{"result":[
                    {"metric":{"digest":"abc","db":"shop"},"value":[1,"9"]},
                    {"metric":{"digest":"abc","db":"crm"},"value":[1,"3"]}]}}"#,
            )),
            ..Default::default()
        };
        let opts = PrometheusMySqlStat {
            schema_label: "db".to_string(),
            ..Default::default()
        };
        let entries = parse_prometheus_response(timings, &counters, &opts).expect("parse");
        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].schema_name.as_deref(), Some("shop"));
        assert_eq!(entries[0].calls, 9, "each schema keeps its own count");
        assert_eq!(entries[1].calls, 3);
    }

    #[test]
    fn call_counts_join_on_digest() {
        // The exporter publishes COUNT_STAR as a series of its own, keyed by
        // digest, exactly like pg_stat_statements keys by queryid.
        let timings = br#"{"data":{"result":[
            {"metric":{"digest_text":"SELECT 1","digest":"abc"},"value":[1,"4.5"]}]}}"#;
        let calls = br#"{"data":{"result":[
            {"metric":{"digest":"abc"},"value":[1,"9"]}]}}"#;
        let entries =
            parse_prometheus_response(timings, &calls_only(calls), &PrometheusMySqlStat::default())
                .expect("parse");
        assert_eq!(entries[0].calls, 9);
        assert!(
            (entries[0].mean_exec_time_ms - 500.0).abs() < 0.001,
            "the mean must follow from the call count, got {}",
            entries[0].mean_exec_time_ms
        );
    }

    #[test]
    fn call_counts_leave_unmatched_statements_alone() {
        let timings = br#"{"data":{"result":[
            {"metric":{"digest_text":"SELECT 1","digest":"abc"},"value":[1,"4.5"]}]}}"#;
        let calls = br#"{"data":{"result":[
            {"metric":{"digest":"zzz"},"value":[1,"9"]}]}}"#;
        let entries =
            parse_prometheus_response(timings, &calls_only(calls), &PrometheusMySqlStat::default())
                .expect("parse");
        assert_eq!(entries[0].calls, 0, "no match means no invented count");
        assert!((entries[0].mean_exec_time_ms - 4500.0).abs() < 0.001);
    }

    fn body(label: &str) -> Vec<u8> {
        format!(
            r#"{{"data":{{"result":[
                {{"metric":{{"schema":"shop","digest":"a1b2","{label}":"SELECT * FROM orders WHERE id = ?"}},
                 "value":[1,"2.5"]}}
            ]}}}}"#
        )
        .into_bytes()
    }

    #[test]
    fn defaults_match_the_mysqld_exporter_collector() {
        let d = PrometheusMySqlStat::default();
        assert_eq!(
            "mysql_perf_schema_events_statements_seconds_total",
            d.series
        );
        assert_eq!("digest_text", d.query_label);
    }

    #[test]
    fn seconds_become_milliseconds_and_the_schema_label_is_kept() {
        let entries = parse_prometheus_response(
            &body("digest_text"),
            &ScrapedCounters::default(),
            &PrometheusMySqlStat::default(),
        )
        .expect("well-formed response");
        assert_eq!(1, entries.len());
        assert!((entries[0].total_exec_time_ms - 2500.0).abs() < f64::EPSILON);
        assert_eq!(Some("shop".to_string()), entries[0].schema_name);
        assert_eq!("SELECT * FROM orders WHERE id = ?", entries[0].query);
    }

    #[test]
    fn query_label_is_overridable() {
        // A recording rule or a hand-written exporter names its own label,
        // so the digest text is not always under `digest_text`.
        let opts = PrometheusMySqlStat::with_overrides(None, Some("statement".to_string()));
        let entries =
            parse_prometheus_response(&body("statement"), &ScrapedCounters::default(), &opts)
                .expect("well-formed response");
        assert_eq!("SELECT * FROM orders WHERE id = ?", entries[0].query);
    }

    #[test]
    fn a_missing_label_falls_back_to_the_opaque_digest() {
        // `digest` still groups rows apart, where "unknown" would collapse
        // every statement into one.
        let entries = parse_prometheus_response(
            &body("other_label"),
            &ScrapedCounters::default(),
            &PrometheusMySqlStat::default(),
        )
        .expect("well-formed response");
        assert_eq!("a1b2", entries[0].query);
    }

    #[test]
    fn the_exporter_carries_no_call_count_so_the_mean_is_the_total() {
        // The series is cumulated seconds only. Reporting a fabricated mean
        // would be worse than repeating the total, which the file export
        // (COUNT_STAR) is there to refine.
        let entries = parse_prometheus_response(
            &body("digest_text"),
            &ScrapedCounters::default(),
            &PrometheusMySqlStat::default(),
        )
        .expect("well-formed response");
        assert_eq!(0, entries[0].calls);
        assert!(
            (entries[0].mean_exec_time_ms - entries[0].total_exec_time_ms).abs() < f64::EPSILON
        );
    }

    #[test]
    fn a_malformed_envelope_is_a_format_error() {
        assert!(
            parse_prometheus_response(
                b"{}",
                &ScrapedCounters::default(),
                &PrometheusMySqlStat::default()
            )
            .is_err()
        );
    }
}

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

    const CSV_HEADER: &str = "SCHEMA_NAME,DIGEST_TEXT,COUNT_STAR,SUM_TIMER_WAIT,AVG_TIMER_WAIT,SUM_ROWS_SENT,SUM_ROWS_EXAMINED";

    fn sample_csv() -> String {
        format!(
            "{CSV_HEADER}\n\
             shop,SELECT * FROM `order_item` WHERE `order_id` = ?,1500,4500500000000,3000000000,1500,45000\n\
             shop,\"SELECT * FROM orders WHERE id IN (?, ?, ?)\",800,2400000000000,3000000000,2400,2400\n\
             NULL,SELECT COUNT ( * ) FROM order_item,50,250000000000,5000000000,50,500000"
        )
    }

    // ----- Format detection -----

    #[test]
    fn detect_format_csv() {
        assert_eq!(
            detect_mysql_stat_format(sample_csv().as_bytes()),
            MySqlStatFormat::Csv
        );
    }

    #[test]
    fn detect_format_json_array() {
        assert_eq!(
            detect_mysql_stat_format(b"[{\"DIGEST_TEXT\": \"SELECT ?\"}]"),
            MySqlStatFormat::Json
        );
    }

    #[test]
    fn detect_format_json_with_whitespace() {
        assert_eq!(
            detect_mysql_stat_format(b"  \n\t [{}]"),
            MySqlStatFormat::Json
        );
    }

    #[test]
    fn detect_format_empty_defaults_csv() {
        assert_eq!(detect_mysql_stat_format(b""), MySqlStatFormat::Csv);
    }

    // ----- CSV parsing -----

    #[test]
    fn parse_csv_basic() {
        let entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        assert_eq!(entries.len(), 3);
        assert_eq!(entries[0].calls, 1500);
        assert_eq!(entries[0].schema_name.as_deref(), Some("shop"));
        assert_eq!(entries[0].rows_sent, 1500);
        assert_eq!(entries[0].rows_examined, 45000);
    }

    #[test]
    fn parse_csv_converts_picoseconds_to_ms() {
        let entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        // 4_500_500_000_000 ps -> 4500.5 ms; 3_000_000_000 ps -> 3.0 ms.
        assert!((entries[0].total_exec_time_ms - 4500.5).abs() < f64::EPSILON);
        assert!((entries[0].mean_exec_time_ms - 3.0).abs() < f64::EPSILON);
    }

    #[test]
    fn parse_csv_huge_timer_exceeding_u64_parses_as_f64() {
        // Aggregated SUM_TIMER_WAIT can exceed u64::MAX (~1.8e19).
        let csv = format!("{CSV_HEADER}\nshop,SELECT ?,1,20000000000000000000,1000000000,1,1");
        let entries = parse_mysql_stat(csv.as_bytes(), 1_048_576).unwrap();
        assert!((entries[0].total_exec_time_ms - 2e10).abs() < 1e-3);
    }

    #[test]
    fn parse_csv_backticked_identifiers_survive_normalization() {
        let entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        assert!(
            entries[0].normalized_template.contains("`order_item`"),
            "backticks must survive: {}",
            entries[0].normalized_template
        );
    }

    #[test]
    fn parse_csv_collapses_in_list() {
        let entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        assert!(
            entries[1].normalized_template.contains("IN (?)"),
            "IN list must collapse: {}",
            entries[1].normalized_template
        );
    }

    #[test]
    fn parse_csv_null_schema_maps_to_none() {
        let entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        assert_eq!(entries[2].schema_name, None);
    }

    #[test]
    fn parse_csv_case_insensitive_headers() {
        let csv = "schema_name,digest_text,count_star,sum_timer_wait,avg_timer_wait\n\
                   shop,SELECT ?,10,1000000000,100000000";
        let entries = parse_mysql_stat(csv.as_bytes(), 1_048_576).unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].calls, 10);
        // Optional columns absent -> defaults.
        assert_eq!(entries[0].rows_sent, 0);
        assert_eq!(entries[0].rows_examined, 0);
    }

    #[test]
    fn parse_csv_missing_required_column() {
        let csv = "DIGEST_TEXT,COUNT_STAR\nSELECT ?,10";
        let result = parse_mysql_stat(csv.as_bytes(), 1_048_576);
        assert_matches!(result, Err(MySqlStatError::MissingColumn(c)) if c == "sum_timer_wait");
    }

    #[test]
    fn parse_csv_quoted_field_with_comma() {
        let csv = format!(
            "{CSV_HEADER}\nshop,\"SELECT `a`, `b` FROM t WHERE id = ?\",5,1000000000,200000000,5,5"
        );
        let entries = parse_mysql_stat(csv.as_bytes(), 1_048_576).unwrap();
        assert!(entries[0].query.contains("`a`, `b`"));
    }

    #[test]
    fn parse_csv_invalid_number_reports_line() {
        let csv = format!("{CSV_HEADER}\nshop,SELECT ?,abc,1000000000,100000000,1,1");
        let result = parse_mysql_stat(csv.as_bytes(), 1_048_576);
        assert_matches!(result, Err(MySqlStatError::CsvParse { line: 2, .. }));
    }

    #[test]
    fn parse_empty_input() {
        assert_matches!(
            parse_mysql_stat(b"", 1_048_576),
            Err(MySqlStatError::EmptyInput)
        );
        assert_matches!(
            parse_mysql_stat(b"   \n  ", 1_048_576),
            Err(MySqlStatError::EmptyInput)
        );
    }

    #[test]
    fn parse_oversized_payload() {
        assert_matches!(
            parse_mysql_stat(&[b'a'; 100], 10),
            Err(MySqlStatError::PayloadTooLarge { size: 100, max: 10 })
        );
    }

    // ----- JSON parsing -----

    #[test]
    fn parse_json_uppercase_keys() {
        let json = r#"[{
            "SCHEMA_NAME": "shop",
            "DIGEST_TEXT": "SELECT * FROM `order_item` WHERE `order_id` = ?",
            "COUNT_STAR": 1500,
            "SUM_TIMER_WAIT": 4500500000000,
            "AVG_TIMER_WAIT": 3000000000,
            "SUM_ROWS_SENT": 1500,
            "SUM_ROWS_EXAMINED": 45000
        }]"#;
        let entries = parse_mysql_stat(json.as_bytes(), 1_048_576).unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].calls, 1500);
        assert!((entries[0].total_exec_time_ms - 4500.5).abs() < f64::EPSILON);
        assert_eq!(entries[0].schema_name.as_deref(), Some("shop"));
    }

    #[test]
    fn parse_json_lowercase_keys() {
        let json = r#"[{
            "digest_text": "SELECT ?",
            "count_star": 10,
            "sum_timer_wait": 1000000000,
            "avg_timer_wait": 100000000
        }]"#;
        let entries = parse_mysql_stat(json.as_bytes(), 1_048_576).unwrap();
        assert_eq!(entries[0].calls, 10);
        assert_eq!(entries[0].schema_name, None);
        assert_eq!(entries[0].rows_examined, 0);
    }

    #[test]
    fn parse_json_null_schema_maps_to_none() {
        let json = r#"[{
            "SCHEMA_NAME": null,
            "DIGEST_TEXT": "SELECT ?",
            "COUNT_STAR": 1,
            "SUM_TIMER_WAIT": 1000000000,
            "AVG_TIMER_WAIT": 1000000000
        }]"#;
        let entries = parse_mysql_stat(json.as_bytes(), 1_048_576).unwrap();
        assert_eq!(entries[0].schema_name, None);
    }

    #[test]
    fn parse_json_empty_array() {
        assert_matches!(
            parse_mysql_stat(b"[]", 1_048_576),
            Err(MySqlStatError::EmptyInput)
        );
    }

    #[test]
    fn parse_json_invalid() {
        assert_matches!(
            parse_mysql_stat(b"[{\"DIGEST_TEXT\": 42}]", 1_048_576),
            Err(MySqlStatError::JsonParse(_))
        );
    }

    // ----- Ranking -----

    #[test]
    fn rank_produces_four_rankings_in_stable_order() {
        let entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        let report = rank_mysql_stat(&entries, 10);
        assert_eq!(report.total_entries, 3);
        let labels: Vec<&str> = report.rankings.iter().map(|r| r.label.as_str()).collect();
        assert_eq!(
            labels,
            [
                "top by total_exec_time",
                "top by calls",
                "top by mean_exec_time",
                "top by rows_examined",
            ]
        );
    }

    #[test]
    fn rank_orders_by_criterion() {
        let entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        let report = rank_mysql_stat(&entries, 10);
        // total time: 4500.5 first; rows_examined: 500_000 (COUNT(*)) first.
        assert_eq!(report.rankings[0].entries[0].calls, 1500);
        assert_eq!(report.rankings[3].entries[0].rows_examined, 500_000);
    }

    #[test]
    fn rank_truncates_to_top_n() {
        let entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        let report = rank_mysql_stat(&entries, 2);
        for ranking in &report.rankings {
            assert_eq!(ranking.entries.len(), 2);
        }
    }

    // ----- Cross-reference -----

    use crate::detect::test_finding_with_template as make_finding;

    #[test]
    fn cross_reference_marks_matching_template() {
        let mut entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        let template = entries[0].normalized_template.clone();
        let findings = vec![make_finding(&template)];
        cross_reference(&mut entries, &findings);
        assert!(entries[0].seen_in_traces);
        assert!(!entries[1].seen_in_traces);
    }

    #[test]
    fn cross_reference_empty_findings() {
        let mut entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        cross_reference(&mut entries, &[]);
        assert!(entries.iter().all(|e| !e.seen_in_traces));
    }

    /// The template set comes from the spans, so a traced statement gets
    /// its marker even when no detector fired on it, and the digest
    /// canonicalization still applies to both sides.
    #[test]
    fn cross_reference_templates_marks_a_traced_statement_without_finding() {
        let csv = format!(
            "{CSV_HEADER}\ncrm,\"SELECT `c` . `name` FROM `customers` `c` WHERE `c` . `id` = ?\",10,1000000000,100000000,10,10"
        );
        let mut entries = parse_mysql_stat(csv.as_bytes(), 1_048_576).unwrap();
        let counts: std::collections::HashMap<String, u64> = [(
            "SELECT c.name FROM customers c WHERE c.id = ?".to_string(),
            3,
        )]
        .into_iter()
        .collect();
        cross_reference_templates(&mut entries, &counts);
        assert!(
            entries[0].seen_in_traces,
            "spaced backticked digest must match the plain trace template"
        );
    }

    #[test]
    fn trace_match_summary_weights_by_calls() {
        let mut entries = parse_mysql_stat(sample_csv().as_bytes(), 1_048_576).unwrap();
        entries[0].seen_in_traces = true;
        let summary = trace_match_summary(&entries);
        assert_eq!(summary.matched_templates, 1);
        assert_eq!(summary.total_templates, entries.len());
        assert_eq!(summary.matched_calls, entries[0].calls);
        let expected_total: u64 = entries.iter().map(|e| e.calls).sum();
        assert_eq!(summary.total_calls, expected_total);
    }

    #[test]
    fn cross_reference_bridges_digest_spacing_and_backticks() {
        // Regression: MySQL DIGEST_TEXT spaces every token and forces
        // backticks; the trace-side template comes from raw application
        // SQL. Exact string equality would silently never match.
        let csv = format!(
            "{CSV_HEADER}\ncrm,\"SELECT `c` . `name` , `o` . `total` FROM `customers` `c` WHERE `c` . `id` = ?\",10,1000000000,100000000,10,10"
        );
        let mut entries = parse_mysql_stat(csv.as_bytes(), 1_048_576).unwrap();
        let findings = vec![make_finding(
            "SELECT c.name, o.total FROM customers c WHERE c.id = ?",
        )];
        cross_reference(&mut entries, &findings);
        assert!(
            entries[0].seen_in_traces,
            "spaced backticked digest must match the plain trace template"
        );
    }

    #[test]
    fn comparison_key_canonicalizes_common_digest_shapes() {
        assert_eq!(
            comparison_key("SELECT `a` . `b` FROM `t` WHERE `a` . `id` IN (?)"),
            comparison_key("select a.b from t where a.id in (?)")
        );
        // Distinct queries stay distinct.
        assert_ne!(
            comparison_key("SELECT `a` FROM `t`"),
            comparison_key("SELECT `b` FROM `t`")
        );
    }

    #[test]
    fn parse_json_null_digest_row_is_skipped() {
        // performance_schema keeps a catch-all row with DIGEST_TEXT NULL
        // once the digest table saturates: skip it, keep the rest.
        let json = r#"[
            {"DIGEST_TEXT": null, "COUNT_STAR": 9999, "SUM_TIMER_WAIT": 1, "AVG_TIMER_WAIT": 1},
            {"DIGEST_TEXT": "SELECT ?", "COUNT_STAR": 10, "SUM_TIMER_WAIT": 1000000000, "AVG_TIMER_WAIT": 100000000}
        ]"#;
        let entries = parse_mysql_stat(json.as_bytes(), 1_048_576).unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].calls, 10);
    }

    #[test]
    fn parse_json_all_null_digests_is_an_error_not_empty_success() {
        // Wrong column names (no DIGEST_TEXT at all) and a fully NULL
        // export must fail loudly, not print "Total entries: 0".
        let json = r#"[
            {"QUERY": "SELECT 1", "COUNT_STAR": 1, "SUM_TIMER_WAIT": 1, "AVG_TIMER_WAIT": 1},
            {"DIGEST_TEXT": null, "COUNT_STAR": 2, "SUM_TIMER_WAIT": 1, "AVG_TIMER_WAIT": 1}
        ]"#;
        assert_matches!(
            parse_mysql_stat(json.as_bytes(), 1_048_576),
            Err(MySqlStatError::JsonParse(_))
        );
    }

    #[test]
    fn comparison_key_bridges_spaced_operators() {
        // MySQL digest text spaces operators the app SQL writes compactly.
        assert_eq!(
            comparison_key("WHERE `a` != ? AND `b` > ?"),
            comparison_key("where a!=? and b>?")
        );
        assert_eq!(
            comparison_key("SELECT `a` + `b` FROM `t`"),
            comparison_key("select a+b from t")
        );
    }

    #[test]
    fn parse_csv_null_digest_row_is_skipped() {
        let csv = format!(
            "{CSV_HEADER}\nshop,NULL,9999,1,1,0,0\nshop,SELECT ?,10,1000000000,100000000,1,1"
        );
        let entries = parse_mysql_stat(csv.as_bytes(), 1_048_576).unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].calls, 10);
    }

    #[test]
    fn parse_csv_backslash_n_schema_maps_to_none() {
        // \N is the INTO OUTFILE-style NULL rendering.
        let csv = format!("{CSV_HEADER}\n\\N,SELECT ?,10,1000000000,100000000,1,1");
        let entries = parse_mysql_stat(csv.as_bytes(), 1_048_576).unwrap();
        assert_eq!(entries[0].schema_name, None);
    }
}