datui-lib 0.4.0

//! Prepare chart data from a LazyFrame: read the chart's columns, then turn them into
//! points, bins or statistics.
//!
//! Every chart reads its rows through [`read_columns`]: up to a row limit of them,
//! spread across the table by the sampler the analysis tools use, so a chart shows the
//! table and not its first rows. What was read comes back as [`RowsRead`], so the chart
//! can say when it shows a sample.

use chrono::{DateTime, Datelike, NaiveDate, NaiveDateTime, NaiveTime};
use color_eyre::Result;
use polars::chunked_array::cast::CastOptions;
use polars::datatypes::{DataType, TimeUnit};
use polars::prelude::*;
use std::f64::consts::PI;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

/// Describes how x-axis numeric values map to temporal types for label formatting.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum XAxisTemporalKind {
    #[default]
    Numeric,
    Date,       // x = days since Unix epoch (f64)
    DatetimeUs, // x = microseconds since epoch
    DatetimeMs,
    DatetimeNs,
    Time, // x = nanoseconds since midnight
}

fn x_axis_temporal_kind(dtype: &DataType) -> XAxisTemporalKind {
    match dtype {
        DataType::Date => XAxisTemporalKind::Date,
        DataType::Datetime(unit, _) => match unit {
            TimeUnit::Nanoseconds => XAxisTemporalKind::DatetimeNs,
            TimeUnit::Microseconds => XAxisTemporalKind::DatetimeUs,
            TimeUnit::Milliseconds => XAxisTemporalKind::DatetimeMs,
        },
        DataType::Time => XAxisTemporalKind::Time,
        _ => XAxisTemporalKind::Numeric,
    }
}

/// Returns the x-axis temporal kind for a column from the schema (for axis label formatting when no data is loaded yet).
pub fn x_axis_temporal_kind_for_column(schema: &Schema, x_column: &str) -> XAxisTemporalKind {
    schema
        .get(x_column)
        .map(x_axis_temporal_kind)
        .unwrap_or(XAxisTemporalKind::Numeric)
}

/// Decimal places past which an axis writes its numbers in scientific notation.
const MAX_AXIS_PLACES: i32 = 6;
/// The most decimals a scientific mantissa takes to tell ticks apart.
const MAX_MANTISSA_PLACES: i32 = 12;
/// Magnitude from which an axis writes its numbers in scientific notation.
const SCIENTIFIC_FROM: f64 = 1e15;

/// What a numeric axis holds: the table's format for its numbers, and whether they are
/// whole (counts, an integer column), which ticks the axis only at whole numbers.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct AxisNumbers {
    pub format: crate::numfmt::NumberFormat,
    pub whole: bool,
}

impl AxisNumbers {
    /// `column`'s numbers as the table prints them; plain when the schema lacks it.
    pub fn column(
        settings: &crate::numfmt::NumberFormatSettings,
        schema: Option<&Schema>,
        column: &str,
    ) -> Self {
        match schema.and_then(|s| s.get(column)) {
            Some(dtype) => Self {
                format: table_number_format(settings, column, dtype),
                whole: dtype.is_integer(),
            },
            None => Self::default(),
        }
    }

    /// Several columns on one axis: the first one's format, whole when every one is.
    pub fn columns(
        settings: &crate::numfmt::NumberFormatSettings,
        schema: Option<&Schema>,
        columns: &[String],
    ) -> Self {
        let mut each = columns.iter().map(|c| Self::column(settings, schema, c));
        let Some(first) = each.next() else {
            return Self::default();
        };
        let whole = first.whole && each.all(|n| n.whole);
        Self { whole, ..first }
    }

    /// Counts, as the table prints a count.
    pub fn count(settings: &crate::numfmt::NumberFormatSettings) -> Self {
        Self {
            format: table_number_format(settings, "Count", &DataType::UInt64),
            whole: true,
        }
    }

    /// A measure of the data such as a density, as the table prints a float.
    pub fn measure(settings: &crate::numfmt::NumberFormatSettings, name: &str) -> Self {
        Self {
            format: table_number_format(settings, name, &DataType::Float64),
            whole: false,
        }
    }

    /// The same numbers, ticked anywhere: on a log scale, or spread by a density.
    pub fn fractional(self) -> Self {
        Self {
            whole: false,
            ..self
        }
    }
}

/// How every tick of one numeric axis is written: one notation and one precision for
/// all of them, chosen from the ticks, in the table's grouping and decimal separator.
/// Chosen per tick, an axis switched to scientific notation partway up.
#[derive(Clone, Debug)]
pub struct AxisFormat {
    format: crate::numfmt::NumberFormat,
    full: Notation,
    /// The shorter form a narrow axis steps down to, when there is one.
    short: Option<Notation>,
    /// Below this a tick is zero: a stepped tick lands a hair off it, which
    /// scientific notation would print as `1.32e-24`.
    zero_below: f64,
}

#[derive(Clone, Copy, Debug, PartialEq)]
enum Notation {
    /// The value in `unit`s to `places` decimals, then `suffix`: `1,234.5`, `12.3k`.
    Fixed {
        places: usize,
        unit: f64,
        suffix: &'static str,
    },
    /// The mantissa to `places` decimals: `1.23e-5`.
    Scientific { places: usize },
    /// Each value in the largest of k, M, G and T it reaches, to the fewest places
    /// up to two that write it exactly: `500`, `2k`, `10M`. A log axis's short form,
    /// whose ticks run across many powers of ten.
    Prefixed,
}

impl AxisFormat {
    /// The format for an axis ticked at `ticks`, in order, holding `numbers`.
    pub fn new(ticks: &[f64], numbers: &AxisNumbers) -> Self {
        let ticks: Vec<f64> = ticks.iter().copied().filter(|v| v.is_finite()).collect();
        let top = ticks.iter().fold(0.0_f64, |top, v| top.max(v.abs()));
        // The closest two ticks, which the labels must still tell apart.
        let gap = ticks
            .windows(2)
            .map(|w| (w[1] - w[0]).abs())
            .filter(|gap| *gap > 0.0)
            .fold(f64::INFINITY, f64::min);
        let places = if top >= SCIENTIFIC_FROM {
            None
        } else if numbers.whole {
            Some(0)
        } else {
            fixed_places(&ticks, top, gap)
        };
        let (full, short) = match places {
            Some(places) => (
                Notation::Fixed {
                    places,
                    unit: 1.0,
                    suffix: "",
                },
                short_notation(&ticks, top, gap),
            ),
            None => {
                // Every mantissa to the places that tell the closest ticks apart.
                let apart = if gap.is_finite() && top > 0.0 {
                    (magnitude(top) - magnitude(gap)).clamp(0, MAX_MANTISSA_PLACES) as usize
                } else {
                    0
                };
                (
                    Notation::Scientific {
                        places: apart.max(2),
                    },
                    Some(Notation::Scientific { places: apart }),
                )
            }
        };
        Self {
            format: numbers.format.clone(),
            full,
            short,
            zero_below: if gap.is_finite() { gap * 1e-9 } else { 0.0 },
        }
    }

    /// The format for a log axis ticked at `ticks`, values before the log: places
    /// enough to write every tick exactly, the 0.1 and 0.25 of a short axis included,
    /// rather than enough to tell the closest two apart, which on a log axis are the
    /// small ones. A 1, 2 or 5 a power of ten up reads `1e18`; the short form names
    /// each tick's own k, M, G or T: `1  10  100  1k  10k`.
    pub fn log(ticks: &[f64], numbers: &AxisNumbers) -> Self {
        let ticks: Vec<f64> = ticks.iter().copied().filter(|v| v.is_finite()).collect();
        let top = ticks.iter().fold(0.0_f64, |top, v| top.max(v.abs()));
        let full = if top >= SCIENTIFIC_FROM {
            Notation::Scientific { places: 0 }
        } else {
            Notation::Fixed {
                places: fewest_places(&ticks, 1.0, 0, MAX_AXIS_PLACES),
                unit: 1.0,
                suffix: "",
            }
        };
        Self {
            format: numbers.format.clone(),
            full,
            short: (1e3..SCIENTIFIC_FROM)
                .contains(&top)
                .then_some(Notation::Prefixed),
            zero_below: 0.0,
        }
    }

    /// The format for an axis from `lo` to `hi` ticked at its ends and halfway, as
    /// [`crate::widgets::axes::AxisSpec::ends_and_middle`] ticks it.
    pub fn ends_and_middle([lo, hi]: [f64; 2], numbers: &AxisNumbers) -> Self {
        Self::new(&[lo, (lo + hi) / 2.0, hi], numbers)
    }

    /// A tick at `level` of detail: 0 the full form, 1 the short one, `None` past the
    /// shortest.
    pub fn label(&self, v: f64, level: usize) -> Option<String> {
        let notation = match level {
            0 => self.full,
            1 => self.short?,
            _ => return None,
        };
        Some(self.write(v, notation))
    }

    fn write(&self, v: f64, notation: Notation) -> String {
        let (places, unit, suffix) = match notation {
            Notation::Scientific { places } => {
                let v = if v.abs() < self.zero_below { 0.0 } else { v };
                return scientific(v, places, self.format.decimal_sep);
            }
            Notation::Fixed { .. } | Notation::Prefixed if !v.is_finite() => {
                return v.to_string();
            }
            Notation::Prefixed => {
                let (unit, suffix) = [(1e12, "T"), (1e9, "G"), (1e6, "M"), (1e3, "k")]
                    .into_iter()
                    .find(|(unit, _)| v.abs() >= *unit)
                    .unwrap_or((1.0, ""));
                let places = fewest_places(&[v], unit, 0, 2);
                (places, unit, suffix)
            }
            Notation::Fixed {
                places,
                unit,
                suffix,
            } => (places, unit, suffix),
        };
        let fixed = crate::numfmt::NumberFormat {
            float_precision: Some(places as u8),
            ..self.format.clone()
        };
        let mut out = String::new();
        fixed.write_f64(v / unit, &mut String::new(), &mut out);
        // A value that rounds to zero is zero: no sign, and no unit to count it in.
        // Checked on the text, since formatting rounds -0.5 to `-0` and `round` to -1.
        if !out.chars().any(|c| matches!(c, '1'..='9')) {
            if unit > 1.0 {
                return "0".to_string();
            }
            out.retain(|c| c != '-');
        }
        out.push_str(suffix);
        out
    }
}

/// Decimal places for `ticks`, whose largest is `top` and closest two `gap` apart:
/// three significant figures of the largest, and enough to tell the closest apart.
/// Fewer when they write every tick exactly, so round ticks read `20`, not `20.0`.
/// `None` for numbers too small to write that way, or ticks too close.
fn fixed_places(ticks: &[f64], top: f64, gap: f64) -> Option<usize> {
    let figures = if top > 0.0 { 2 - magnitude(top) } else { 0 };
    let apart = if gap.is_finite() { -magnitude(gap) } else { 0 };
    let places = figures.max(apart).max(0);
    (places <= MAX_AXIS_PLACES).then(|| fewest_places(ticks, 1.0, apart.max(0), places))
}

/// The fewest places from `least` to `most` that write every one of `ticks`, counted
/// in `unit`s, exactly; `most` when none do.
fn fewest_places(ticks: &[f64], unit: f64, least: i32, most: i32) -> usize {
    let exact = |places: i32| {
        ticks.iter().all(|v| {
            let scaled = v / unit * 10f64.powi(places);
            // Ticks are stepped in floating point: 0.1 * 3 is 0.30000000000000004.
            (scaled - scaled.round()).abs() <= 1e-9 * scaled.abs().max(1.0)
        })
    };
    (least..most).find(|&p| exact(p)).unwrap_or(most) as usize
}

/// The short form of `ticks`, whose largest is `top`: counted in the k, M, G or T of
/// the largest, to two significant figures of it and places enough to tell ticks
/// `gap` apart, at most two, or fewer as [`fixed_places`] takes them. `None` below a
/// thousand, where there is no shorter form.
fn short_notation(ticks: &[f64], top: f64, gap: f64) -> Option<Notation> {
    let (unit, suffix) = [(1e12, "T"), (1e9, "G"), (1e6, "M"), (1e3, "k")]
        .into_iter()
        .find(|(unit, _)| top >= *unit)?;
    let figures = 1 - magnitude(top / unit);
    let apart = if gap.is_finite() {
        -magnitude(gap / unit)
    } else {
        0
    };
    let most = figures.max(apart).clamp(0, 2);
    Some(Notation::Fixed {
        places: fewest_places(ticks, unit, apart.clamp(0, most), most),
        unit,
        suffix,
    })
}

/// The power of ten `v` is counted in: 1 for 12.3, -2 for 0.05. A hair under a power
/// of ten counts as it, since ticks come of floating-point arithmetic: 1000.3 less
/// 1000.2 is 0.09999... and not a tenth.
fn magnitude(v: f64) -> i32 {
    (v.log10() + 1e-9).floor() as i32
}

/// `v` in scientific notation, its mantissa to `places` decimals.
fn scientific(v: f64, places: usize, decimal_sep: char) -> String {
    // No `-0.00e0`.
    let v = if v == 0.0 { 0.0 } else { v };
    let text = format!("{v:.places$e}");
    if decimal_sep == '.' {
        text
    } else {
        text.replacen('.', decimal_sep.encode_utf8(&mut [0; 4]), 1)
    }
}

/// The format the table prints `column` in, plain where it prints it unformatted.
pub fn table_number_format(
    settings: &crate::numfmt::NumberFormatSettings,
    column: &str,
    dtype: &DataType,
) -> crate::numfmt::NumberFormat {
    match settings.formatter_for(column, dtype) {
        crate::numfmt::CellFormatter::Number(format) => format,
        crate::numfmt::CellFormatter::Passthrough => crate::numfmt::NumberFormat::PLAIN,
    }
}

/// An x value as the date and time it stands for, when `kind` is a date or datetime.
pub(crate) fn x_datetime(v: f64, kind: XAxisTemporalKind) -> Option<NaiveDateTime> {
    const UNIX_EPOCH_CE_DAYS: i32 = 719_163;
    match kind {
        XAxisTemporalKind::Date => NaiveDate::from_num_days_from_ce_opt(
            UNIX_EPOCH_CE_DAYS.saturating_add(v.trunc() as i32),
        )
        .map(|d| d.and_time(NaiveTime::MIN)),
        XAxisTemporalKind::DatetimeUs => {
            DateTime::from_timestamp_micros(v.trunc() as i64).map(|dt| dt.naive_utc())
        }
        XAxisTemporalKind::DatetimeMs => {
            DateTime::from_timestamp_millis(v.trunc() as i64).map(|dt| dt.naive_utc())
        }
        XAxisTemporalKind::DatetimeNs => {
            DateTime::from_timestamp_millis((v.trunc() as i64) / 1_000_000).map(|dt| dt.naive_utc())
        }
        XAxisTemporalKind::Numeric | XAxisTemporalKind::Time => None,
    }
}

/// An x value as a time of day, when `kind` is a time.
pub(crate) fn x_time(v: f64) -> Option<NaiveTime> {
    let nsecs = v.trunc() as u64;
    NaiveTime::from_num_seconds_from_midnight_opt(
        (nsecs / 1_000_000_000) as u32,
        (nsecs % 1_000_000_000) as u32,
    )
}

/// An x tick at `level` of detail, 0 the fullest, or `None` past the shortest form.
/// A narrow axis steps down until its labels fit: a date to year-month and then the
/// year, or to month-day when both ends of the axis, `bounds`, fall in one year; a
/// datetime first to its date, or to the minute when the axis spans one day; a time
/// to the minute. A number, or a time past what it can stand for, is written in
/// `numbers`.
pub fn x_axis_label_at(
    v: f64,
    kind: XAxisTemporalKind,
    bounds: (f64, f64),
    level: usize,
    numbers: &AxisFormat,
) -> Option<String> {
    if kind == XAxisTemporalKind::Numeric {
        return numbers.label(v, level);
    }
    if kind == XAxisTemporalKind::Time {
        let pattern = ["%H:%M:%S", "%H:%M"].get(level)?;
        return Some(match x_time(v) {
            Some(t) => t.format(pattern).to_string(),
            None => numbers.label(v, level)?,
        });
    }
    let Some(at) = x_datetime(v, kind) else {
        return numbers.label(v, level);
    };
    let ends = x_datetime(bounds.0, kind).zip(x_datetime(bounds.1, kind));
    let one_day = ends.is_some_and(|(a, b)| a.date() == b.date());
    let one_year = ends.is_some_and(|(a, b)| a.year() == b.year());
    let dates: &[&str] = if one_year {
        &["%Y-%m-%d", "%m-%d"]
    } else {
        &["%Y-%m-%d", "%Y-%m", "%Y"]
    };
    let patterns: Vec<&str> = if kind == XAxisTemporalKind::Date {
        dates.to_vec()
    } else if one_day {
        vec!["%Y-%m-%d %H:%M", "%H:%M"]
    } else {
        std::iter::once("%Y-%m-%d %H:%M")
            .chain(dates.iter().copied())
            .collect()
    };
    patterns.get(level).map(|p| at.format(p).to_string())
}

/// How a chart reads its rows.
#[derive(Clone, Debug)]
pub struct ChartSampling {
    /// Rows to read; `None` reads every row.
    pub limit: Option<usize>,
    /// The view's row count when the table already knows it, which saves a count.
    pub known_total: Option<usize>,
    /// The shared analysis seed, so a chart and Describe draw alike.
    pub seed: u64,
    pub streaming: bool,
    /// Whether the view may be read whole, twice, for a line's envelope: not a scan
    /// of an object store in place, where the sample reads a few row groups and the
    /// envelope would download everything twice. See [`prepare_chart_data`].
    pub full_passes: bool,
    /// The rows already read from this view.
    pub held: HeldRows,
    /// Set once nobody wants the result: a streamed count stops at its next batch.
    pub cancel: Arc<AtomicBool>,
}

impl ChartSampling {
    /// Up to `limit` rows, with the analysis tools' default seed.
    pub fn rows(limit: Option<usize>) -> Self {
        Self {
            limit,
            known_total: None,
            seed: crate::sampling::Sample::default().seed,
            streaming: false,
            full_passes: true,
            held: HeldRows::default(),
            cancel: Arc::default(),
        }
    }
}

/// The rows a chart last read from one view. Another bin count, range, bandwidth or
/// chart over columns already read draws from them instead of reading the table
/// again, and every chart of the view describes the same sample. Shared with the
/// worker that reads; whoever owns the view starts a new one when the view changes.
#[derive(Clone, Default)]
pub struct HeldRows(Arc<Mutex<Holding>>);

#[derive(Default)]
struct Holding {
    rows: Option<Held>,
    /// Rows per category, from a count of the whole view: exact whatever the sample
    /// size, so another order or size draws from them rather than counting again.
    counts: Vec<HeldCounts>,
}

struct Held {
    limit: Option<usize>,
    seed: u64,
    df: DataFrame,
    rows: RowsRead,
}

struct HeldCounts {
    category: String,
    counted: Counted,
}

impl std::fmt::Debug for HeldRows {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("HeldRows")
    }
}

/// What a chart read: the rows the table has, and how many of them were sampled when
/// that was fewer.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct RowsRead {
    pub total_rows: usize,
    pub sample_size: Option<usize>,
    /// A line chart over more rows than its sample size draws each of this many steps
    /// along X as its lowest and highest value instead of sampling: see
    /// [`prepare_chart_data`].
    pub envelope_steps: Option<usize>,
    /// The seed the sample was drawn with, when it is a sample.
    pub seed: Option<u64>,
}

/// Which values a histogram, box plot or KDE draws. Outliers far from the body squash
/// it into a bin or two; a percentile range leaves them out and says how many.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ValueRange {
    #[default]
    All,
    /// The 1st to the 99th percentile.
    Percentile1To99,
}

impl ValueRange {
    pub const ALL: [Self; 2] = [Self::All, Self::Percentile1To99];

    pub fn label(self) -> &'static str {
        match self {
            Self::All => "All",
            Self::Percentile1To99 => "p1-p99",
        }
    }

    fn quantiles(self) -> Option<(f64, f64)> {
        match self {
            Self::All => None,
            Self::Percentile1To99 => Some((0.01, 0.99)),
        }
    }
}

/// A range that left values out: which, and how many.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Clipped {
    pub range: ValueRange,
    pub outside: usize,
}

/// What a chart says under the plot about its input, one line each: that it is a
/// sample, with the seed that draws it again, and how many values a range left out.
/// Empty when it shows every row and every value. `middot` joins the seed on.
pub fn chart_notes(rows: &RowsRead, clipped: Option<&Clipped>, middot: &str) -> Vec<String> {
    let mut notes = Vec::new();
    if let Some(steps) = rows.envelope_steps {
        notes.push(format!(
            "min and max of {} rows in {} steps",
            crate::discover::format_rows(rows.total_rows),
            crate::numfmt::group_chrome(steps)
        ));
    }
    if let Some(n) = rows.sample_size {
        let mut note = format!(
            "sample of {} of {} rows",
            crate::numfmt::group_chrome(n),
            crate::discover::format_rows(rows.total_rows)
        );
        if let Some(seed) = rows.seed {
            note.push_str(&format!(" {middot} seed {seed}"));
        }
        notes.push(note);
    }
    if let Some(clipped) = clipped {
        let noun = if clipped.outside == 1 {
            "value"
        } else {
            "values"
        };
        notes.push(format!(
            "{} {noun} outside {}",
            crate::numfmt::group_chrome(clipped.outside),
            clipped.range.label()
        ));
    }
    notes
}

/// Read `columns` (each once, however often named) through the analysis sampler: every
/// row up to the limit, and past it a seeded sample spread across the table — runs of
/// one Parquet or IPC file, or one streamed pass over anything else — never its head.
///
/// Rows already held for the same size and seed are used as they are when they have
/// the columns. Otherwise the read takes the held columns along, so going back to one
/// does not read again.
fn read_columns(
    lf: &LazyFrame,
    columns: &[&str],
    sampling: &ChartSampling,
) -> Result<(DataFrame, RowsRead)> {
    let mut unique: Vec<PlSmallStr> = Vec::with_capacity(columns.len());
    for c in columns {
        if !unique.iter().any(|u| u == c) {
            unique.push((*c).into());
        }
    }
    let mut holding = sampling.held.0.lock().unwrap_or_else(|e| e.into_inner());
    if let Some(h) = holding
        .rows
        .as_ref()
        .filter(|h| h.limit == sampling.limit && h.seed == sampling.seed)
    {
        if unique.iter().all(|c| h.df.column(c).is_ok()) {
            return Ok((h.df.select(unique.iter().cloned())?, h.rows));
        }
        for c in h.df.get_column_names() {
            if !unique.contains(c) {
                unique.push(c.clone());
            }
        }
    }
    let lf = lf
        .clone()
        .select(unique.iter().map(|c| col(c.clone())).collect::<Vec<_>>());
    let read = crate::statistics::analysis_rows(
        &lf,
        sampling.limit,
        sampling.known_total,
        sampling.seed,
        sampling.streaming,
    )?;
    let rows = RowsRead {
        total_rows: read.total_rows,
        sample_size: read.sample_size,
        envelope_steps: None,
        seed: read.sample_size.map(|_| sampling.seed),
    };
    holding.rows = Some(Held {
        limit: sampling.limit,
        seed: sampling.seed,
        df: read.df.clone(),
        rows,
    });
    Ok((read.df, rows))
}

/// A column's values as `f64`, one per row; null, NaN and infinities are `None`.
fn f64_values(df: &DataFrame, column: &str) -> Result<Vec<Option<f64>>> {
    let cast = df.column(column)?.cast(&DataType::Float64)?;
    Ok(cast
        .f64()?
        .iter()
        .map(|v| v.filter(|v| v.is_finite()))
        .collect())
}

/// X as `f64`, one per row: numbers as they are, temporal types as their ordinal (see
/// [`XAxisTemporalKind`]).
fn x_values(df: &DataFrame, column: &str, dtype: &DataType) -> Result<Vec<Option<f64>>> {
    match dtype {
        DataType::Datetime(_, _) | DataType::Date | DataType::Time => {
            let ordinal = df.column(column)?.cast(&DataType::Int64)?;
            Ok(ordinal.i64()?.iter().map(|v| v.map(|v| v as f64)).collect())
        }
        _ => f64_values(df, column),
    }
}

/// Result of loading only the x column: min/max for axis bounds and temporal kind.
pub struct ChartXRangeResult {
    pub x_min: f64,
    pub x_max: f64,
    pub x_axis_kind: XAxisTemporalKind,
    pub rows: RowsRead,
}

/// Loads only the x column and returns its min/max (for axis display when no y is selected).
pub fn prepare_chart_x_range(
    lf: &LazyFrame,
    schema: &Schema,
    x_column: &str,
    sampling: &ChartSampling,
) -> Result<ChartXRangeResult> {
    let x_dtype = schema
        .get(x_column)
        .ok_or_else(|| color_eyre::eyre::eyre!("x column '{}' not in schema", x_column))?;
    let x_axis_kind = x_axis_temporal_kind(x_dtype);
    let (df, rows) = read_columns(lf, &[x_column], sampling)?;
    let (x_min, x_max) = x_values(&df, x_column, x_dtype)?
        .into_iter()
        .flatten()
        .fold((f64::INFINITY, f64::NEG_INFINITY), |(lo, hi), x| {
            (lo.min(x), hi.max(x))
        });
    let (x_min, x_max) = if x_max >= x_min {
        (x_min, x_max)
    } else {
        (0.0, 1.0)
    };
    Ok(ChartXRangeResult {
        x_min,
        x_max,
        x_axis_kind,
        rows,
    })
}

/// Result of preparing chart data: series points and x-axis kind for label formatting.
pub struct ChartDataResult {
    /// One per y column, in X order.
    pub series: Vec<Vec<(f64, f64)>>,
    /// Per series, the indices of `series` where a line starts again after a gap: a row
    /// whose y is null is left out of that series alone, and a line does not bridge it.
    pub breaks: Vec<Vec<usize>>,
    pub x_axis_kind: XAxisTemporalKind,
    pub rows: RowsRead,
}

/// A series split at its breaks: the runs a line joins.
pub fn segments<'a>(points: &'a [(f64, f64)], breaks: &[usize]) -> Vec<&'a [(f64, f64)]> {
    let mut out = Vec::with_capacity(breaks.len() + 1);
    let mut start = 0;
    for &b in breaks {
        if b > start && b <= points.len() {
            out.push(&points[start..b]);
            start = b;
        }
    }
    if start < points.len() {
        out.push(&points[start..]);
    }
    out
}

/// Histogram bin (center and count).
#[derive(Clone, Debug)]
pub struct HistogramBin {
    pub center: f64,
    pub count: f64,
}

/// One group's bins of a histogram split by a color: its count (or share) per bin,
/// on the bins of the whole.
#[derive(Clone, Debug)]
pub struct HistogramGroup {
    pub name: String,
    pub counts: Vec<f64>,
}

/// Histogram data for a single column.
#[derive(Clone, Debug)]
pub struct HistogramData {
    pub column: String,
    /// Every row's bins; with `share`, each bin's share of the rows.
    pub bins: Vec<HistogramBin>,
    /// Per color group, on the same bins: drawn as outlines over one another.
    pub groups: Vec<HistogramGroup>,
    /// The last group is Other: every value of the color without a group of its own.
    pub other: bool,
    /// Each bin is a share of its group's rows (or of all rows), not a count.
    pub share: bool,
    pub x_min: f64,
    pub x_max: f64,
    pub max_count: f64,
    pub rows: RowsRead,
    pub clipped: Option<Clipped>,
}

/// KDE series and bounds.
#[derive(Clone, Debug)]
pub struct KdeSeries {
    pub name: String,
    pub points: Vec<(f64, f64)>,
}

#[derive(Clone, Debug)]
pub struct KdeData {
    pub series: Vec<KdeSeries>,
    /// The last series is Other: every value of the color without a series of its own.
    pub other: bool,
    pub x_min: f64,
    pub x_max: f64,
    pub y_max: f64,
    pub rows: RowsRead,
    pub clipped: Option<Clipped>,
}

/// Box plot stats for a column.
#[derive(Clone, Debug)]
pub struct BoxPlotStats {
    pub name: String,
    pub min: f64,
    pub q1: f64,
    pub median: f64,
    pub q3: f64,
    pub max: f64,
}

#[derive(Clone, Debug)]
pub struct BoxPlotData {
    pub stats: Vec<BoxPlotStats>,
    pub y_min: f64,
    pub y_max: f64,
    pub rows: RowsRead,
    pub clipped: Option<Clipped>,
    /// One box per category: how many categories there are, of which the largest
    /// have a box. 0 for a box per column.
    pub of: usize,
}

/// Heatmap data for two numeric columns.
#[derive(Clone, Debug)]
pub struct HeatmapData {
    pub x_column: String,
    pub y_column: String,
    pub x_min: f64,
    pub x_max: f64,
    pub y_min: f64,
    pub y_max: f64,
    pub x_bins: usize,
    pub y_bins: usize,
    pub counts: Vec<Vec<f64>>,
    pub max_count: f64,
    pub rows: RowsRead,
}

/// Prepares XY series from the current LazyFrame. X is cast to f64 (temporal types as
/// ordinal). Nulls are dropped per series: a null X drops the row, a null Y drops that
/// series' point and breaks its line. Points come in X order, so a line runs left to
/// right whatever order the rows are in; ties keep table order.
///
/// With `envelope`, a view of more rows than the sample size is not sampled: X is cut
/// into half that many steps and each step draws its lowest and highest Y. A random
/// sample of a waveform or any long series joins points far apart and misses its peaks;
/// the envelope keeps every peak, as many steps as a plot has columns. It reads the
/// view twice, so only where [`ChartSampling::full_passes`] allows; both passes stop
/// when [`ChartSampling::cancel`] is set.
pub fn prepare_chart_data(
    lf: &LazyFrame,
    schema: &Schema,
    x_column: &str,
    y_columns: &[String],
    sampling: &ChartSampling,
    envelope: bool,
) -> Result<ChartDataResult> {
    if y_columns.is_empty() {
        return Ok(ChartDataResult {
            series: Vec::new(),
            breaks: Vec::new(),
            x_axis_kind: XAxisTemporalKind::Numeric,
            rows: RowsRead::default(),
        });
    }

    let x_dtype = schema
        .get(x_column)
        .ok_or_else(|| color_eyre::eyre::eyre!("x column '{}' not in schema", x_column))?;
    let x_axis_kind = x_axis_temporal_kind(x_dtype);

    let mut counted = None;
    if envelope
        && sampling.full_passes
        && let Some(limit) = sampling.limit.filter(|&n| n > 0)
        && sampling.known_total.is_none_or(|n| n > limit)
    {
        match envelope_series(lf, x_column, x_dtype, y_columns, limit, sampling)? {
            Envelope::Drawn {
                series,
                breaks,
                rows,
                steps,
            } => {
                return Ok(ChartDataResult {
                    series,
                    breaks,
                    x_axis_kind,
                    rows: RowsRead {
                        total_rows: rows,
                        sample_size: None,
                        envelope_steps: Some(steps),
                        seed: None,
                    },
                });
            }
            // The first pass counted the view: the sample takes it whole.
            Envelope::Fits(rows) => counted = Some(rows),
        }
    }
    let counted_sampling;
    let sampling = match counted {
        Some(rows) => {
            counted_sampling = ChartSampling {
                known_total: Some(rows),
                ..sampling.clone()
            };
            &counted_sampling
        }
        None => sampling,
    };

    let mut columns = vec![x_column];
    columns.extend(y_columns.iter().map(String::as_str));
    let (df, rows) = read_columns(lf, &columns, sampling)?;

    let mut order: Vec<(f64, usize)> = x_values(&df, x_column, x_dtype)?
        .into_iter()
        .enumerate()
        .filter_map(|(i, x)| x.map(|x| (x, i)))
        .collect();
    // Stable, so rows sharing an X keep their table order.
    order.sort_by(|a, b| a.0.total_cmp(&b.0));

    let mut series = Vec::with_capacity(y_columns.len());
    let mut breaks = Vec::with_capacity(y_columns.len());
    for y_column in y_columns {
        let ys = f64_values(&df, y_column)?;
        let mut points = Vec::with_capacity(order.len());
        let mut starts = Vec::new();
        let mut gap = false;
        for &(x, i) in &order {
            match ys[i] {
                Some(y) => {
                    if gap && !points.is_empty() {
                        starts.push(points.len());
                    }
                    gap = false;
                    points.push((x, y));
                }
                None => gap = true,
            }
        }
        series.push(points);
        breaks.push(starts);
    }

    Ok(ChartDataResult {
        series,
        breaks,
        x_axis_kind,
        rows,
    })
}

/// What [`envelope_series`] found.
enum Envelope {
    /// Per Y column, its points and where its line breaks, from `rows` rows in
    /// `steps` steps.
    Drawn {
        series: Vec<Vec<(f64, f64)>>,
        breaks: Vec<Vec<usize>>,
        rows: usize,
        steps: usize,
    },
    /// The view is no more rows than the sample size, this many: no envelope.
    Fits(usize),
}

/// What a pass stopped by [`until_cancelled`] fails with.
const ENVELOPE_CANCELLED: &str = "chart cancelled";

/// `e`, failing the query once `cancel` is set: a streamed pass stops at its next
/// morsel rather than reading on for a chart nobody waits for.
fn until_cancelled(e: Expr, cancel: &Arc<AtomicBool>) -> Expr {
    let cancel = Arc::clone(cancel);
    e.map(
        move |c: Column| {
            polars_ensure!(!cancel.load(Ordering::Relaxed), ComputeError: ENVELOPE_CANCELLED);
            Ok(c)
        },
        |_, field| Ok(field.clone()),
    )
}

/// Collect a pass of the envelope, streamed whatever the setting: it holds a few
/// numbers per step. A pass stopped by `cancel` is an error that says so.
fn envelope_pass(lf: LazyFrame, cancel: &Arc<AtomicBool>) -> Result<DataFrame> {
    crate::statistics::collect_lazy(lf, true).map_err(|e| {
        if cancel.load(Ordering::Relaxed) {
            color_eyre::eyre::eyre!(ENVELOPE_CANCELLED)
        } else {
            e.into()
        }
    })
}

/// Per Y column, half `limit` steps along X, each its lowest and highest finite Y at
/// the step's lowest X, in X order. Two streamed passes over the view: the rows and
/// X's bounds, then one group per step. A row with no X is left out whole; a step
/// with rows where a series has no value breaks its line.
fn envelope_series(
    lf: &LazyFrame,
    x_column: &str,
    x_dtype: &DataType,
    y_columns: &[String],
    limit: usize,
    sampling: &ChartSampling,
) -> Result<Envelope> {
    let cancel = &sampling.cancel;
    // Temporal X as its ordinal, as `x_values` reads it.
    let x = match x_dtype {
        DataType::Datetime(_, _) | DataType::Date | DataType::Time | DataType::Duration(_) => {
            col(x_column).cast(DataType::Int64).cast(DataType::Float64)
        }
        _ => col(x_column).cast(DataType::Float64),
    };
    // Not finite is null, so the aggregations below are a plain min and max, which
    // stream; a filter inside them does not.
    let finite = |e: Expr| {
        when(e.clone().is_finite())
            .then(e)
            .otherwise(lit(NULL).cast(DataType::Float64))
    };
    let x = finite(until_cancelled(x, cancel)).alias("__x");
    let bounds = envelope_pass(
        lf.clone().select([
            len().alias("rows"),
            x.clone().min().alias("lo"),
            x.clone().max().alias("hi"),
        ]),
        cancel,
    )?;
    let rows = bounds
        .column("rows")?
        .cast(&DataType::UInt64)?
        .u64()?
        .get(0)
        .unwrap_or(0) as usize;
    if rows <= limit {
        return Ok(Envelope::Fits(rows));
    }
    let steps = (limit / 2).max(1);
    let n = y_columns.len();
    let drawn = |series, breaks| Envelope::Drawn {
        series,
        breaks,
        rows,
        steps,
    };
    let bound = |name: &str| -> Result<Option<f64>> { Ok(bounds.column(name)?.f64()?.get(0)) };
    let (Some(lo), Some(hi)) = (bound("lo")?, bound("hi")?) else {
        return Ok(drawn(vec![Vec::new(); n], vec![Vec::new(); n]));
    };
    let per_x = if hi > lo {
        steps as f64 / (hi - lo)
    } else {
        0.0
    };
    let lf = lf
        .clone()
        .select(
            std::iter::once(x)
                .chain(y_columns.iter().enumerate().map(|(i, y)| {
                    finite(col(y.as_str()).cast(DataType::Float64)).alias(format!("__y{i}"))
                }))
                .collect::<Vec<_>>(),
        )
        // A row filter, not a filter inside the select: that would leave X shorter than
        // the Y columns beside it.
        .filter(col("__x").is_not_null());
    let step = ((col("__x") - lit(lo)) * lit(per_x))
        .floor()
        .cast(DataType::Int64)
        .clip(lit(0i64), lit(steps as i64 - 1))
        .alias("__step");
    let mut aggs = vec![col("__x").min()];
    for i in 0..n {
        let y = col(format!("__y{i}"));
        aggs.push(y.clone().min().alias(format!("__lo{i}")));
        aggs.push(y.max().alias(format!("__hi{i}")));
    }
    let df = envelope_pass(
        lf.group_by([step])
            .agg(aggs)
            .sort(["__step"], Default::default()),
        cancel,
    )?;
    let xs = df.column("__x")?.f64()?.clone();
    let mut series = Vec::with_capacity(n);
    let mut breaks = Vec::with_capacity(n);
    for i in 0..n {
        let lows = df.column(&format!("__lo{i}"))?.f64()?.clone();
        let highs = df.column(&format!("__hi{i}"))?.f64()?.clone();
        let mut points = Vec::with_capacity(xs.len() * 2);
        let mut starts = Vec::new();
        let mut gap = false;
        for ((x, low), high) in xs.iter().zip(lows.iter()).zip(highs.iter()) {
            let (Some(x), Some(low), Some(high)) = (x, low, high) else {
                gap = true;
                continue;
            };
            if gap && !points.is_empty() {
                starts.push(points.len());
            }
            gap = false;
            points.push((x, low));
            if high != low {
                points.push((x, high));
            }
        }
        series.push(points);
        breaks.push(starts);
    }
    Ok(drawn(series, breaks))
}

/// Each column's finite values, read in one pass; nulls are dropped per column.
fn read_values(
    lf: &LazyFrame,
    columns: &[&str],
    sampling: &ChartSampling,
) -> Result<(Vec<Vec<f64>>, RowsRead)> {
    let (df, rows) = read_columns(lf, columns, sampling)?;
    let values = columns
        .iter()
        .map(|c| Ok(f64_values(&df, c)?.into_iter().flatten().collect()))
        .collect::<Result<Vec<Vec<f64>>>>()?;
    Ok((values, rows))
}

/// Sort `values` and keep those inside `range`; returns how many were left out.
fn sort_and_clip(values: &mut Vec<f64>, range: ValueRange) -> usize {
    values.sort_by(f64::total_cmp);
    let Some((low, high)) = range.quantiles() else {
        return 0;
    };
    if values.is_empty() {
        return 0;
    }
    let (low, high) = (quantile(values, low), quantile(values, high));
    let before = values.len();
    values.retain(|v| (low..=high).contains(v));
    before - values.len()
}

fn clipped(range: ValueRange, outside: usize) -> Option<Clipped> {
    (range != ValueRange::All).then_some(Clipped { range, outside })
}

/// Prepare histogram data for a numeric column.
pub fn prepare_histogram_data(
    lf: &LazyFrame,
    column: &str,
    bins: usize,
    range: ValueRange,
    sampling: &ChartSampling,
) -> Result<HistogramData> {
    prepare_histogram_by(lf, column, bins, range, false, None, sampling)
}

/// Prepare a histogram of `column`, split by `color` into groups on the same bins
/// when given. With `share`, a bin is its share of its group's rows (of every row,
/// unsplit), so groups of different sizes compare. The range is the whole
/// column's, so every group is clipped alike.
pub fn prepare_histogram_by(
    lf: &LazyFrame,
    column: &str,
    bins: usize,
    range: ValueRange,
    share: bool,
    color: Option<ColorSplit<'_>>,
    sampling: &ChartSampling,
) -> Result<HistogramData> {
    let (values, rows) = read_split(lf, column, color, sampling)?;
    let mut all: Vec<f64> = values.iter().map(|(v, _)| *v).collect();
    let outside = sort_and_clip(&mut all, range);
    let clipped = clipped(range, outside);
    let mut data = HistogramData {
        column: column.to_string(),
        bins: Vec::new(),
        groups: Vec::new(),
        other: false,
        share,
        x_min: 0.0,
        x_max: 1.0,
        max_count: 0.0,
        rows,
        clipped,
    };
    let (Some(&lo), Some(&hi)) = (all.first(), all.last()) else {
        return Ok(data);
    };
    let span = hi - lo;
    let bin_count = if span <= f64::EPSILON { 1 } else { bins.max(1) };
    let bin_width = if span <= f64::EPSILON {
        1.0
    } else {
        span / bin_count as f64
    };
    (data.x_min, data.x_max) = if span <= f64::EPSILON {
        (lo - 0.5, hi + 0.5)
    } else {
        (lo, hi)
    };
    let bin_of = |v: f64| {
        if span <= f64::EPSILON {
            0
        } else {
            (((v - lo) / bin_width).floor().max(0.0) as usize).min(bin_count - 1)
        }
    };
    let groups = color.map_or(0, |c| c.series());
    let mut total = vec![0.0_f64; bin_count];
    let mut by_group = vec![vec![0.0_f64; bin_count]; groups];
    for (v, group) in values {
        // The range is the whole view's; the bins count the groups drawn.
        if !(lo..=hi).contains(&v) || (color.is_some() && group.is_none()) {
            continue;
        }
        let bin = bin_of(v);
        total[bin] += 1.0;
        if let Some(g) = group {
            by_group[g][bin] += 1.0;
        }
    }
    let as_share = |counts: &mut Vec<f64>| {
        let n: f64 = counts.iter().sum();
        if share && n > 0.0 {
            counts.iter_mut().for_each(|c| *c /= n);
        }
    };
    as_share(&mut total);
    by_group.iter_mut().for_each(as_share);
    let center = |i: usize| {
        if span <= f64::EPSILON {
            lo
        } else {
            lo + (i as f64 + 0.5) * bin_width
        }
    };
    data.bins = total
        .iter()
        .enumerate()
        .map(|(i, &count)| HistogramBin {
            center: center(i),
            count,
        })
        .collect();
    let max = |counts: &[f64]| counts.iter().copied().fold(0.0_f64, f64::max);
    if let Some(color) = color {
        data.groups = color
            .names()
            .into_iter()
            .zip(by_group)
            .map(|(name, counts)| HistogramGroup { name, counts })
            .collect();
        data.other = color.other;
        data.max_count = data
            .groups
            .iter()
            .map(|g| max(&g.counts))
            .fold(0.0, f64::max);
    } else {
        data.max_count = max(&total);
    }
    Ok(data)
}

fn quantile(sorted: &[f64], q: f64) -> f64 {
    if sorted.is_empty() {
        return 0.0;
    }
    let n = sorted.len();
    if n == 1 {
        return sorted[0];
    }
    let pos = q.clamp(0.0, 1.0) * (n as f64 - 1.0);
    let idx = pos.floor() as usize;
    let next = pos.ceil() as usize;
    if idx == next {
        sorted[idx]
    } else {
        let lower = sorted[idx];
        let upper = sorted[next];
        let weight = pos - idx as f64;
        lower + (upper - lower) * weight
    }
}

/// The five numbers of sorted `values`, or `None` when there are none.
fn box_stats(name: String, values: &[f64]) -> Option<BoxPlotStats> {
    let (min, max) = (*values.first()?, *values.last()?);
    Some(BoxPlotStats {
        name,
        min,
        q1: quantile(values, 0.25),
        median: quantile(values, 0.5),
        q3: quantile(values, 0.75),
        max,
    })
}

/// Box plot data from stats, its bounds taken from them.
fn box_data(stats: Vec<BoxPlotStats>, rows: RowsRead, clipped: Option<Clipped>) -> BoxPlotData {
    let mut y_min = stats.iter().map(|s| s.min).fold(f64::INFINITY, f64::min);
    let mut y_max = stats
        .iter()
        .map(|s| s.max)
        .fold(f64::NEG_INFINITY, f64::max);
    if stats.is_empty() {
        (y_min, y_max) = (0.0, 1.0);
    } else if y_max <= y_min {
        y_max = y_min + 1.0;
    }
    BoxPlotData {
        stats,
        y_min,
        y_max,
        rows,
        clipped,
        of: 0,
    }
}

/// Prepare box plot stats for one or more numeric columns. Uses a single read for all columns.
pub fn prepare_box_plot_data<T: AsRef<str>>(
    lf: &LazyFrame,
    columns: &[T],
    range: ValueRange,
    sampling: &ChartSampling,
) -> Result<BoxPlotData> {
    let col_refs: Vec<&str> = columns.iter().map(|c| c.as_ref()).collect();
    let (columns_values, rows) = if col_refs.is_empty() {
        (Vec::new(), RowsRead::default())
    } else {
        read_values(lf, &col_refs, sampling)?
    };
    let mut stats = Vec::new();
    let mut outside = 0;
    for (column, mut values) in col_refs.iter().zip(columns_values) {
        outside += sort_and_clip(&mut values, range);
        stats.extend(box_stats((*column).to_string(), &values));
    }
    Ok(box_data(stats, rows, clipped(range, outside)))
}

/// Prepare one box of `column` per group of `by`: a box per category. The range is
/// each group's own, as each box describes its group.
pub fn prepare_box_by(
    lf: &LazyFrame,
    column: &str,
    by: ColorSplit<'_>,
    range: ValueRange,
    sampling: &ChartSampling,
) -> Result<BoxPlotData> {
    let (values, rows) = read_split(lf, column, Some(by), sampling)?;
    let mut groups = vec![Vec::new(); by.groups.len()];
    for (v, group) in values {
        if let Some(g) = group {
            groups[g].push(v);
        }
    }
    let mut outside = 0;
    let mut stats = Vec::new();
    for (name, mut values) in by.groups.iter().zip(groups) {
        outside += sort_and_clip(&mut values, range);
        stats.extend(box_stats(group_label(name), &values));
    }
    Ok(box_data(stats, rows, clipped(range, outside)))
}

fn kde_bandwidth(values: &[f64]) -> f64 {
    if values.len() <= 1 {
        return 1.0;
    }
    let n = values.len() as f64;
    let mean = values.iter().sum::<f64>() / n;
    let var = values.iter().map(|v| (v - mean).powi(2)).sum::<f64>() / n;
    let std = var.sqrt();
    if std <= f64::EPSILON {
        return 1.0;
    }
    1.06 * std * n.powf(-0.2)
}

/// The density of sorted `values` at 200 points, from three bandwidths below the
/// least to three above the greatest; `None` when there are none.
fn kde_series(name: String, values: &[f64], bandwidth_factor: f64) -> Option<KdeSeries> {
    let (min, max) = (*values.first()?, *values.last()?);
    let bandwidth = (kde_bandwidth(values) * bandwidth_factor).max(f64::EPSILON);
    let x_start = min - 3.0 * bandwidth;
    let x_end = max + 3.0 * bandwidth;
    let samples = 200_usize;
    let step = (x_end - x_start) / (samples.saturating_sub(1).max(1) as f64);
    let inv = 1.0 / ((values.len() as f64) * bandwidth * (2.0 * PI).sqrt());
    let points = (0..samples)
        .map(|i| {
            let x = x_start + i as f64 * step;
            let sum: f64 = values
                .iter()
                .map(|&v| {
                    let u = (x - v) / bandwidth;
                    (-0.5 * u * u).exp()
                })
                .sum();
            (x, inv * sum)
        })
        .collect();
    Some(KdeSeries { name, points })
}

/// KDE data from its series, its bounds taken from them.
fn kde_data(series: Vec<KdeSeries>, rows: RowsRead, clipped: Option<Clipped>) -> KdeData {
    let points = || series.iter().flat_map(|s| s.points.iter());
    let mut x_min = points().map(|p| p.0).fold(f64::INFINITY, f64::min);
    let mut x_max = points().map(|p| p.0).fold(f64::NEG_INFINITY, f64::max);
    let mut y_max = points().map(|p| p.1).fold(f64::NEG_INFINITY, f64::max);
    if series.is_empty() {
        (x_min, x_max, y_max) = (0.0, 1.0, 1.0);
    }
    if x_max <= x_min {
        x_max = x_min + 1.0;
    }
    if y_max <= 0.0 {
        y_max = 1.0;
    }
    KdeData {
        series,
        other: false,
        x_min,
        x_max,
        y_max,
        rows,
        clipped,
    }
}

/// Prepare KDE data for one or more numeric columns. Uses a single read for all columns.
pub fn prepare_kde_data<T: AsRef<str>>(
    lf: &LazyFrame,
    columns: &[T],
    bandwidth_factor: f64,
    range: ValueRange,
    sampling: &ChartSampling,
) -> Result<KdeData> {
    let col_refs: Vec<&str> = columns.iter().map(|c| c.as_ref()).collect();
    let (columns_values, rows) = if col_refs.is_empty() {
        (Vec::new(), RowsRead::default())
    } else {
        read_values(lf, &col_refs, sampling)?
    };
    let mut series = Vec::new();
    let mut outside = 0;
    for (column, mut values) in col_refs.iter().zip(columns_values) {
        outside += sort_and_clip(&mut values, range);
        series.extend(kde_series((*column).to_string(), &values, bandwidth_factor));
    }
    Ok(kde_data(series, rows, clipped(range, outside)))
}

/// Prepare one density curve of `column` per group of `color`. The range is the
/// whole column's, so every curve is clipped alike.
pub fn prepare_kde_by(
    lf: &LazyFrame,
    column: &str,
    bandwidth_factor: f64,
    range: ValueRange,
    color: ColorSplit<'_>,
    sampling: &ChartSampling,
) -> Result<KdeData> {
    let (values, rows) = read_split(lf, column, Some(color), sampling)?;
    let mut all: Vec<f64> = values.iter().map(|(v, _)| *v).collect();
    let outside = sort_and_clip(&mut all, range);
    let (lo, hi) = match (all.first(), all.last()) {
        (Some(&lo), Some(&hi)) => (lo, hi),
        _ => (f64::INFINITY, f64::NEG_INFINITY),
    };
    let mut groups = vec![Vec::new(); color.series()];
    for (v, group) in values {
        if let Some(g) = group
            && (lo..=hi).contains(&v)
        {
            groups[g].push(v);
        }
    }
    let last = color.series().saturating_sub(1);
    let mut other = false;
    let series = color
        .names()
        .into_iter()
        .zip(groups)
        .enumerate()
        .filter_map(|(i, (name, mut values))| {
            values.sort_by(f64::total_cmp);
            let series = kde_series(name, &values, bandwidth_factor)?;
            other = color.other && i == last;
            Some(series)
        })
        .collect();
    Ok(KdeData {
        other,
        ..kde_data(series, rows, clipped(range, outside))
    })
}

/// Prepare heatmap data for two numeric columns. A row counts when both are present.
pub fn prepare_heatmap_data(
    lf: &LazyFrame,
    x_column: &str,
    y_column: &str,
    bins: usize,
    sampling: &ChartSampling,
) -> Result<HeatmapData> {
    let (df, rows) = read_columns(lf, &[x_column, y_column], sampling)?;
    let pairs: Vec<(f64, f64)> = f64_values(&df, x_column)?
        .into_iter()
        .zip(f64_values(&df, y_column)?)
        .filter_map(|(x, y)| Some((x?, y?)))
        .collect();
    let x_bins = bins.max(1);
    let y_bins = bins.max(1);
    if pairs.is_empty() {
        return Ok(HeatmapData {
            x_column: x_column.to_string(),
            y_column: y_column.to_string(),
            x_min: 0.0,
            x_max: 1.0,
            y_min: 0.0,
            y_max: 1.0,
            x_bins,
            y_bins,
            counts: vec![vec![0.0; x_bins]; y_bins],
            max_count: 0.0,
            rows,
        });
    }
    let mut x_min = f64::INFINITY;
    let mut x_max = f64::NEG_INFINITY;
    let mut y_min = f64::INFINITY;
    let mut y_max = f64::NEG_INFINITY;
    for (x, y) in &pairs {
        x_min = x_min.min(*x);
        x_max = x_max.max(*x);
        y_min = y_min.min(*y);
        y_max = y_max.max(*y);
    }
    if x_max <= x_min {
        x_max = x_min + 1.0;
    }
    if y_max <= y_min {
        y_max = y_min + 1.0;
    }
    let mut counts = vec![vec![0.0_f64; x_bins]; y_bins];
    let x_range = x_max - x_min;
    let y_range = y_max - y_min;
    for (x, y) in pairs {
        let xi =
            (((x - x_min) / x_range * x_bins as f64).floor().max(0.0) as usize).min(x_bins - 1);
        let yi =
            (((y - y_min) / y_range * y_bins as f64).floor().max(0.0) as usize).min(y_bins - 1);
        counts[yi][xi] += 1.0;
    }
    let max_count = counts
        .iter()
        .flat_map(|row| row.iter())
        .cloned()
        .fold(0.0_f64, f64::max);
    Ok(HeatmapData {
        x_column: x_column.to_string(),
        y_column: y_column.to_string(),
        x_min,
        x_max,
        y_min,
        y_max,
        x_bins,
        y_bins,
        counts,
        max_count,
        rows,
    })
}

/// The order of a bar chart's bars, top to bottom.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BarOrder {
    /// Largest value first.
    #[default]
    Value,
    /// The category's own order: text A to Z, numbers ascending, false before true.
    Label,
}

impl BarOrder {
    pub const ALL: [Self; 2] = [Self::Value, Self::Label];

    pub fn label(self) -> &'static str {
        match self {
            Self::Value => "Value",
            Self::Label => "Label",
        }
    }
}

/// Most bars a bar chart prepares; the rest are counted, not drawn. More than a tall
/// terminal shows, few enough that an export stays legible.
pub const BAR_CAP: usize = 100;

/// Column types a bar chart takes as its category: text, categories, booleans and
/// integers. A float or a timestamp is a measure, not a category.
pub fn is_category_dtype(dtype: &DataType) -> bool {
    matches!(
        dtype,
        DataType::String | DataType::Categorical(_, _) | DataType::Enum(_, _) | DataType::Boolean
    ) || dtype.is_integer()
}

/// What sets a bar's length: a numeric column, one row per category, or how many rows
/// each category has.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum BarValue {
    Count,
    Column(String),
}

impl BarValue {
    pub fn label(&self) -> &str {
        match self {
            Self::Count => "Count",
            Self::Column(column) => column,
        }
    }
}

/// Categories a count keeps before it stops: past this the column is an identifier,
/// not a category, and a count per value would hold as much as the table.
pub const COUNT_CATEGORY_CAP: usize = 100_000;

/// One bar: its category (`None` for a null category) and its value. Split by a
/// color, a bar is a row of bars, one per group (`None` where a group has no rows),
/// and its value is their total.
#[derive(Clone, Debug, PartialEq)]
pub struct Bar {
    pub label: Option<String>,
    pub value: f64,
    pub by_group: Vec<Option<f64>>,
}

/// A bar chart: one bar per category, in order, up to [`BAR_CAP`].
#[derive(Clone, Debug)]
pub struct BarData {
    pub category: String,
    pub value_column: String,
    pub bars: Vec<Bar>,
    /// Categories past the cap, not in `bars`.
    pub more: usize,
    /// Categories left out because their value is null.
    pub no_value: usize,
    pub rows: RowsRead,
    /// The value column's type: an integer column prints whole numbers.
    pub value_dtype: DataType,
    /// Rows counted, when a count read past the sample size: the counts are of every
    /// row of the view, and the note says so.
    pub counted: Option<usize>,
    /// The color groups each bar is split into, in color order; empty when not split.
    pub groups: Vec<String>,
    /// The last group is Other: every value of the color without a group of its own.
    pub other: bool,
    /// What an aggregate read, said under the plot: `all 336,776 rows`.
    pub rows_note: Option<String>,
}

impl BarData {
    /// The format the table prints the value column in, or plain where it shows the
    /// column unformatted.
    pub fn value_format(
        &self,
        settings: &crate::numfmt::NumberFormatSettings,
    ) -> crate::numfmt::NumberFormat {
        table_number_format(settings, &self.value_column, &self.value_dtype)
    }

    /// Each bar's value as the table prints the value column.
    pub fn value_labels(&self, settings: &crate::numfmt::NumberFormatSettings) -> Vec<String> {
        self.labels_in(&self.value_format(settings))
    }

    /// Each bar's value in `format`: whole for an integer column or a count.
    pub fn labels_in(&self, format: &crate::numfmt::NumberFormat) -> Vec<String> {
        let integer = self.value_dtype.is_integer();
        self.bars
            .iter()
            .map(|b| format_bar_value(b.value, integer, format))
            .collect()
    }
}

/// Format a bar's value for the label beside it in `format`: an integer column's whole,
/// anything else to the format's decimal places or two, so every bar shows the same
/// number of them. Values too large, or too small to show in those places, go to
/// scientific notation.
pub fn format_bar_value(v: f64, integer: bool, format: &crate::numfmt::NumberFormat) -> String {
    let places = format.float_precision.unwrap_or(2);
    let smallest = 0.5 * 10f64.powi(-i32::from(places));
    if !v.is_finite() || v.abs() >= 1e15 || (!integer && v != 0.0 && v.abs() < smallest) {
        return scientific(v, 2, format.decimal_sep);
    }
    let mut out = String::new();
    if integer {
        format.write_i64(v as i64, &mut out);
    } else {
        let fixed = crate::numfmt::NumberFormat {
            float_precision: Some(places),
            ..format.clone()
        };
        fixed.write_f64(v, &mut String::new(), &mut out);
    }
    out
}

/// A column name as SQL reads it: bare when it is a plain lowercase identifier, quoted
/// otherwise.
fn sql_ident(name: &str) -> String {
    let plain = name
        .chars()
        .next()
        .is_some_and(|c| c.is_ascii_lowercase() || c == '_')
        && name
            .chars()
            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_');
    if plain {
        name.to_string()
    } else {
        format!("\"{}\"", name.replace('"', "\"\""))
    }
}

/// Prepare a bar chart: one bar per row, the category from `category`, the length from
/// `value`. The chart takes a grouped result — one row per category — and refuses a
/// category that repeats rather than guess how to combine its rows. A null category is
/// a bar of its own; a null value leaves its category out, counted in `no_value`.
pub fn prepare_bar_data(
    lf: &LazyFrame,
    category: &str,
    value: &str,
    order: BarOrder,
    cap: usize,
    sampling: &ChartSampling,
) -> Result<BarData> {
    let (df, rows) = read_columns(lf, &[category, value], sampling)?;
    let categories = df.column(category)?.as_materialized_series().clone();
    let labels_series = crate::past_calendar::cast_text(&categories, CastOptions::NonStrict)?;
    let labels: Vec<Option<&str>> = labels_series.str()?.iter().collect();

    let mut seen: std::collections::HashMap<Option<&str>, usize> =
        std::collections::HashMap::with_capacity(labels.len());
    for label in &labels {
        *seen.entry(*label).or_default() += 1;
    }
    if seen.len() < labels.len() {
        let read = match rows.sample_size {
            Some(n) => format!("a sample of {} rows", crate::numfmt::group_chrome(n)),
            None => format!("{} rows", crate::numfmt::group_chrome(labels.len())),
        };
        let (c, v) = (sql_ident(category), sql_ident(value));
        // The q form only where it reads the names as they are.
        let q = if c == category && v == value {
            format!(" (or select avg {value} by {category})")
        } else {
            String::new()
        };
        return Err(color_eyre::eyre::eyre!(
            "{category} repeats: {} categories in {read}. A bar takes one row per category, \
             so group first: SELECT {c}, AVG({v}) FROM df GROUP BY {c}{q}, or choose Count \
             for the rows per category",
            crate::numfmt::group_chrome(seen.len()),
        ));
    }

    let values = f64_values(&df, value)?;
    let (bars, more, no_value) = order_bars(&categories, &labels, &values, order, cap);
    Ok(BarData {
        category: category.to_string(),
        value_column: value.to_string(),
        bars,
        more,
        no_value,
        rows,
        value_dtype: df.column(value)?.dtype().clone(),
        counted: None,
        groups: Vec::new(),
        other: false,
        rows_note: None,
    })
}

/// A bar per row in `order`, up to `cap`: the bars, how many are past the cap, and how
/// many rows had no value. Equal values keep row order.
fn order_bars(
    categories: &Series,
    labels: &[Option<&str>],
    values: &[Option<f64>],
    order: BarOrder,
    cap: usize,
) -> (Vec<Bar>, usize, usize) {
    let row_order: Vec<usize> = match order {
        BarOrder::Value => (0..labels.len()).collect(),
        BarOrder::Label => label_order(categories),
    };
    let mut no_value = 0;
    let mut bars: Vec<Bar> = row_order
        .into_iter()
        .filter_map(|i| match values[i] {
            Some(value) => Some(Bar {
                label: labels[i].map(str::to_string),
                value,
                by_group: Vec::new(),
            }),
            None => {
                no_value += 1;
                None
            }
        })
        .collect();
    if order == BarOrder::Value {
        // Stable: equal values keep row order.
        bars.sort_by(|a, b| b.value.total_cmp(&a.value));
    }
    let more = bars.len().saturating_sub(cap);
    bars.truncate(cap);
    (bars, more, no_value)
}

/// Rows in the category's own order: text A to Z, numbers ascending, an enum in its
/// order, nulls last.
fn label_order(categories: &Series) -> Vec<usize> {
    categories
        .arg_sort(
            SortOptions::default()
                .with_nulls_last(true)
                .with_maintain_order(true),
        )
        .iter()
        .flatten()
        .map(|i| i as usize)
        .collect()
}

/// Prepare a bar chart of how many rows each category has. The counts are exact: the
/// whole view is counted, whatever the sample size, in one streamed pass that keeps a
/// count per category and no rows. When the rows held are the whole view, those are
/// counted instead. Past [`COUNT_CATEGORY_CAP`] categories the count stops, and says
/// so rather than drawing part of the view as the whole.
pub fn prepare_bar_counts(
    lf: &LazyFrame,
    category: &str,
    order: BarOrder,
    cap: usize,
    sampling: &ChartSampling,
) -> Result<BarData> {
    count_bars(lf, category, order, cap, COUNT_CATEGORY_CAP, sampling)
}

fn count_bars(
    lf: &LazyFrame,
    category: &str,
    order: BarOrder,
    cap: usize,
    max_categories: usize,
    sampling: &ChartSampling,
) -> Result<BarData> {
    let counted = match held_counts(sampling, category, max_categories)? {
        Some(counted) => counted,
        None => {
            // A view the sample size takes whole is read as the other charts read it,
            // so they draw from the same rows after; a larger one is streamed.
            let fits = sampling
                .limit
                .zip(sampling.known_total)
                .is_some_and(|(n, total)| total <= n);
            let whole = if fits {
                let (df, rows) = read_columns(lf, &[category], sampling)?;
                rows.sample_size.is_none().then_some(df)
            } else {
                None
            };
            let counted = match whole {
                Some(df) => count_frame(&df, category, max_categories)?,
                None => stream_counts(lf, category, max_categories, &sampling.cancel)?,
            };
            hold_counts(sampling, category, &counted);
            counted
        }
    };
    let (counts, total) = match counted {
        Counted::All { counts, rows } => (counts, rows),
        Counted::TooMany => {
            return Err(color_eyre::eyre::eyre!(
                "more than {} categories of {category}: counting stopped. Count by a \
                 column with fewer values",
                crate::numfmt::group_chrome(max_categories)
            ));
        }
    };
    let data = |bars, more| BarData {
        category: category.to_string(),
        value_column: "count".to_string(),
        bars,
        more,
        no_value: 0,
        rows: RowsRead {
            total_rows: total,
            sample_size: None,
            envelope_steps: None,
            seed: None,
        },
        value_dtype: DataType::UInt64,
        counted: sampling.limit.is_some_and(|n| total > n).then_some(total),
        groups: Vec::new(),
        other: false,
        rows_note: None,
    };
    let Some(counts) = counts else {
        return Ok(data(Vec::new(), 0));
    };
    // Label order first, so categories with equal counts come A to Z.
    let by_label: Vec<IdxSize> = label_order(counts.column(category)?.as_materialized_series())
        .into_iter()
        .map(|i| i as IdxSize)
        .collect();
    let counts = counts.take(&IdxCa::from_vec("order".into(), by_label))?;
    let categories = counts.column(category)?.as_materialized_series().clone();
    let labels_series = crate::past_calendar::cast_text(&categories, CastOptions::NonStrict)?;
    let labels: Vec<Option<&str>> = labels_series.str()?.iter().collect();
    let values: Vec<Option<f64>> = counts
        .column(COUNT_COLUMN)?
        .u64()?
        .iter()
        .map(|n| n.map(|n| n as f64))
        .collect();
    let (bars, more, _) = order_bars(&categories, &labels, &values, order, cap);
    Ok(data(bars, more))
}

/// The view's counts of `category` without reading it: counted before, or counted now
/// from rows held that are the whole view.
fn held_counts(
    sampling: &ChartSampling,
    category: &str,
    max_categories: usize,
) -> Result<Option<Counted>> {
    let holding = sampling.held.0.lock().unwrap_or_else(|e| e.into_inner());
    if let Some(held) = holding.counts.iter().find(|h| h.category == category) {
        return Ok(Some(held.counted.clone()));
    }
    let Some(whole) = holding
        .rows
        .as_ref()
        .filter(|h| h.rows.sample_size.is_none() && h.df.column(category).is_ok())
    else {
        return Ok(None);
    };
    let counted = count_frame(&whole.df, category, max_categories)?;
    drop(holding);
    hold_counts(sampling, category, &counted);
    Ok(Some(counted))
}

fn hold_counts(sampling: &ChartSampling, category: &str, counted: &Counted) {
    let mut holding = sampling.held.0.lock().unwrap_or_else(|e| e.into_inner());
    holding.counts.retain(|h| h.category != category);
    if holding.counts.len() >= HELD_COUNTS {
        holding.counts.remove(0);
    }
    holding.counts.push(HeldCounts {
        category: category.to_string(),
        counted: counted.clone(),
    });
}

/// Count the categories of rows already in memory.
pub(crate) fn count_frame(
    df: &DataFrame,
    category: &str,
    max_categories: usize,
) -> Result<Counted> {
    let mut tally = Tally::new(category, max_categories);
    tally.observe(&df.select([category])?)?;
    Ok(tally.finish()?)
}

/// What a count found: one row per category with its rows in [`COUNT_COLUMN`] (none
/// when the view has no rows), and the rows counted; or more categories than it keeps.
#[derive(Clone)]
pub(crate) enum Counted {
    All {
        counts: Option<DataFrame>,
        rows: usize,
    },
    TooMany,
}

pub(crate) const COUNT_COLUMN: &str = "__datui_bar_count";

/// Counts held per view: enough to go back and forth between a few categories, each
/// up to [`COUNT_CATEGORY_CAP`] rows.
const HELD_COUNTS: usize = 4;

/// Rows piled up unmerged before a merge, at least: a run of new categories costs a
/// merge now and then rather than one per batch.
const MERGE_AFTER: usize = 1 << 16;

/// Rows per category, added up batch by batch. Holds one row per category and the
/// batches since the last merge, never the rows themselves.
pub(crate) struct Tally {
    category: PlSmallStr,
    max: usize,
    counts: Option<DataFrame>,
    /// Rows of `counts` as last merged: one per category.
    merged: usize,
    rows: usize,
    too_many: bool,
    /// Stopped before the end of the view because nobody wants the count.
    cancelled: bool,
}

impl Tally {
    pub(crate) fn new(category: &str, max: usize) -> Self {
        Self {
            category: category.into(),
            max,
            counts: None,
            merged: 0,
            rows: 0,
            too_many: false,
            cancelled: false,
        }
    }

    /// Count a batch of the category column. True once there are more categories than
    /// the count keeps, which stops the read.
    pub(crate) fn observe(&mut self, batch: &DataFrame) -> PolarsResult<bool> {
        if self.too_many {
            return Ok(true);
        }
        self.rows += batch.height();
        let part = group_counts(batch, &self.category, false)?;
        let mut counts = match self.counts.take() {
            Some(mut counts) => {
                counts.vstack_mut(&part)?;
                counts
            }
            None => part,
        };
        if counts.height() - self.merged >= self.merged.max(MERGE_AFTER) {
            counts = group_counts(&counts, &self.category, true)?;
            self.merged = counts.height();
            self.too_many = self.merged > self.max;
        }
        self.counts = Some(counts);
        Ok(self.too_many)
    }

    pub(crate) fn finish(self) -> PolarsResult<Counted> {
        let counts = match self.counts {
            Some(counts) => Some(group_counts(&counts, &self.category, true)?),
            None => None,
        };
        if self.too_many || counts.as_ref().is_some_and(|c| c.height() > self.max) {
            return Ok(Counted::TooMany);
        }
        Ok(Counted::All {
            counts,
            rows: self.rows,
        })
    }
}

/// One row per category of `df` with how many rows it stands for: its rows, or when
/// `summed` the counts they already carry, added up. A null category is one of them.
fn group_counts(df: &DataFrame, category: &str, summed: bool) -> PolarsResult<DataFrame> {
    let by = df.group_by([category])?;
    let groups = by.get_groups();
    let counts: Vec<u64> = if summed {
        let carried: Vec<u64> = df
            .column(COUNT_COLUMN)?
            .u64()?
            .into_no_null_iter()
            .collect();
        groups
            .iter()
            .map(|group| match group {
                GroupsIndicator::Idx((_, rows)) => rows.iter().map(|&i| carried[i as usize]).sum(),
                GroupsIndicator::Slice([first, len]) => {
                    carried[first as usize..(first + len) as usize].iter().sum()
                }
            })
            .collect()
    } else {
        groups.iter().map(|group| group.len() as u64).collect()
    };
    let mut columns = by.keys();
    columns.push(Column::new(COUNT_COLUMN.into(), counts));
    DataFrame::new_infer_height(columns)
}

/// Count the view's categories in one streamed pass, stopping past `max` of them, or
/// as soon as `cancel` is set: a pass over a large table can take minutes, and the
/// next chart waits for it.
fn stream_counts(
    lf: &LazyFrame,
    category: &str,
    max: usize,
    cancel: &Arc<AtomicBool>,
) -> Result<Counted> {
    let state = Arc::new(Mutex::new(Tally::new(category, max)));
    let callback_state = Arc::clone(&state);
    let callback_cancel = Arc::clone(cancel);
    let sink = lf.clone().select([col(category)]).sink_batches(
        PlanCallback::new(move |batch: DataFrame| {
            let mut tally = callback_state
                .lock()
                .map_err(|_| PolarsError::ComputeError("count lock failed".into()))?;
            if callback_cancel.load(Ordering::Relaxed) {
                tally.cancelled = true;
                return Ok(true);
            }
            tally.observe(&batch)
        }),
        false,
        None,
    )?;
    // Streaming whatever the setting: a count per category is all this holds.
    crate::statistics::collect_lazy(sink, true)?;
    let tally = std::mem::replace(
        &mut *state.lock().unwrap_or_else(|e| e.into_inner()),
        Tally::new(category, max),
    );
    // Part of the view counted is not a count of it. A pass that finished before it
    // was told to stop is whole, and kept.
    if tally.cancelled {
        return Err(color_eyre::eyre::eyre!("count cancelled"));
    }
    Ok(tally.finish()?)
}

// ----- Color: one series per value -----

/// A column a chart is split by, and the values given a group each, in color order.
/// `None` is the rows with no value. With `other`, every other value's rows make one
/// more group after them, [`OTHER`].
#[derive(Clone, Copy, Debug)]
pub struct ColorSplit<'a> {
    pub column: &'a str,
    pub groups: &'a [Option<String>],
    pub other: bool,
}

/// The name of the group of every value not given one of its own.
pub const OTHER: &str = "Other";

impl ColorSplit<'_> {
    /// How many groups the rows fall in: the values', and Other.
    pub fn series(&self) -> usize {
        self.groups.len() + usize::from(self.other)
    }

    /// Each group's name, as a legend writes it: Other last.
    pub fn names(&self) -> Vec<String> {
        let mut names: Vec<String> = self.groups.iter().map(group_label).collect();
        if self.other {
            names.push(OTHER.to_string());
        }
        names
    }
}

/// A group's name as a legend writes it.
pub fn group_label(value: &Option<String>) -> String {
    value.clone().unwrap_or_else(|| "null".to_string())
}

/// Each row's group: its place among `split.groups`, Other's after them, or `None`
/// for a value that has none. Values are compared as text, as the value picker lists
/// them.
fn row_groups(df: &DataFrame, split: ColorSplit<'_>) -> Result<Vec<Option<usize>>> {
    let series = df.column(split.column)?.as_materialized_series();
    let text = crate::past_calendar::cast_text(series, CastOptions::NonStrict)?;
    let index: std::collections::HashMap<Option<&str>, usize> = split
        .groups
        .iter()
        .enumerate()
        .map(|(i, g)| (g.as_deref(), i))
        .collect();
    let other = split.other.then_some(split.groups.len());
    Ok(text
        .str()?
        .iter()
        .map(|v| index.get(&v).copied().or(other))
        .collect())
}

/// Values with the group each is in, and what was read for them.
type SplitValues = (Vec<(f64, Option<usize>)>, RowsRead);

/// `column`'s finite values as read for a chart, each with its group when split
/// (`None` for a row of a value no group has, which still counts toward a range).
fn read_split(
    lf: &LazyFrame,
    column: &str,
    split: Option<ColorSplit<'_>>,
    sampling: &ChartSampling,
) -> Result<SplitValues> {
    let mut columns = vec![column];
    if let Some(split) = split {
        columns.push(split.column);
    }
    let (df, rows) = read_columns(lf, &columns, sampling)?;
    let values = f64_values(&df, column)?;
    let groups = split.map(|s| row_groups(&df, s)).transpose()?;
    let out = values
        .into_iter()
        .enumerate()
        .filter_map(|(i, v)| {
            let v = v?;
            Some((v, groups.as_ref().and_then(|groups| groups[i])))
        })
        .collect();
    Ok((out, rows))
}

/// A column's values with the rows holding each, most rows first (equal counts in
/// the column's order), counted over the whole view.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct ValueRows {
    pub values: Vec<(Option<String>, u64)>,
    /// Rows counted.
    pub rows: usize,
}

/// Count `column`'s values over the whole view, or take the count held for it: one
/// streamed pass that keeps a count per value, the one a bar chart of counts makes.
pub fn value_rows(lf: &LazyFrame, column: &str, sampling: &ChartSampling) -> Result<ValueRows> {
    let counted = match held_counts(sampling, column, COUNT_CATEGORY_CAP)? {
        Some(counted) => counted,
        None => {
            let counted = stream_counts(lf, column, COUNT_CATEGORY_CAP, &sampling.cancel)?;
            hold_counts(sampling, column, &counted);
            counted
        }
    };
    let (counts, rows) = match counted {
        Counted::All { counts, rows } => (counts, rows),
        Counted::TooMany => {
            return Err(color_eyre::eyre::eyre!(
                "more than {} values of {column}: choose a column with fewer",
                crate::numfmt::group_chrome(COUNT_CATEGORY_CAP)
            ));
        }
    };
    let Some(counts) = counts else {
        return Ok(ValueRows {
            values: Vec::new(),
            rows,
        });
    };
    let by_label: Vec<IdxSize> = label_order(counts.column(column)?.as_materialized_series())
        .into_iter()
        .map(|i| i as IdxSize)
        .collect();
    let counts = counts.take(&IdxCa::from_vec("order".into(), by_label))?;
    let labels = crate::past_calendar::cast_text(
        counts.column(column)?.as_materialized_series(),
        CastOptions::NonStrict,
    )?;
    let mut values: Vec<(Option<String>, u64)> = labels
        .str()?
        .iter()
        .zip(counts.column(COUNT_COLUMN)?.u64()?.iter())
        .map(|(label, n)| (label.map(str::to_string), n.unwrap_or(0)))
        .collect();
    // Stable, so equal counts keep the column's order.
    values.sort_by_key(|v| std::cmp::Reverse(v.1));
    Ok(ValueRows { values, rows })
}

/// The groups a color makes: the values picked, in the order picked, or else the
/// largest by rows; at most `most`, one per series color.
pub fn color_groups(
    rows: &ValueRows,
    picked: &[Option<String>],
    most: usize,
) -> Vec<Option<String>> {
    if !picked.is_empty() {
        return picked.iter().take(most).cloned().collect();
    }
    rows.values
        .iter()
        .take(most)
        .map(|(value, _)| value.clone())
        .collect()
}

/// In a plan: each row's group as its place among `split.groups` (UInt32), Other's
/// after them, null for a value that has none.
fn group_expr(split: ColorSplit<'_>) -> Expr {
    let text = crate::past_calendar::text_expr(col(split.column), CastOptions::NonStrict);
    let mut out = match split.other {
        true => lit(split.groups.len() as u32).cast(DataType::UInt32),
        false => lit(NULL).cast(DataType::UInt32),
    };
    for (i, group) in split.groups.iter().enumerate().rev() {
        let matches = match group {
            Some(value) => text.clone().eq(lit(value.clone())),
            None => col(split.column).is_null(),
        };
        out = when(matches).then(lit(i as u32)).otherwise(out);
    }
    out
}

/// Series of a line or scatter chart, each named: a Y column's or a color group's.
#[derive(Clone, Debug, Default)]
pub struct GroupedSeries {
    pub names: Vec<String>,
    pub series: Vec<Vec<(f64, f64)>>,
    /// Per series, where its line starts again after a gap.
    pub breaks: Vec<Vec<usize>>,
    pub x_axis_kind: XAxisTemporalKind,
    pub rows: RowsRead,
    /// The last series is Other: every value of the color without a series of its own.
    pub other: bool,
}

/// A line or scatter chart of `y` split by `color`, from the rows a chart samples:
/// one series per group, in X order, each breaking where its Y has no value.
pub fn prepare_xy_by(
    lf: &LazyFrame,
    schema: &Schema,
    x: &str,
    y: &str,
    color: ColorSplit<'_>,
    sampling: &ChartSampling,
) -> Result<GroupedSeries> {
    let x_dtype = schema
        .get(x)
        .ok_or_else(|| color_eyre::eyre::eyre!("x column '{}' not in schema", x))?;
    let (df, rows) = read_columns(lf, &[x, y, color.column], sampling)?;
    let xs = x_values(&df, x, x_dtype)?;
    let ys = f64_values(&df, y)?;
    let groups = row_groups(&df, color)?;
    let mut order: Vec<(f64, usize)> = xs
        .into_iter()
        .enumerate()
        .filter_map(|(i, x)| x.map(|x| (x, i)))
        .collect();
    order.sort_by(|a, b| a.0.total_cmp(&b.0));
    let n = color.series();
    let mut series = vec![Vec::new(); n];
    let mut breaks = vec![Vec::new(); n];
    let mut gap = vec![false; n];
    for (x, i) in order {
        let Some(g) = groups[i] else { continue };
        match ys[i] {
            Some(y) => {
                if gap[g] && !series[g].is_empty() {
                    breaks[g].push(series[g].len());
                }
                gap[g] = false;
                series[g].push((x, y));
            }
            None => gap[g] = true,
        }
    }
    Ok(GroupedSeries {
        names: color.names(),
        series,
        breaks,
        x_axis_kind: x_axis_temporal_kind(x_dtype),
        rows,
        other: color.other,
    })
}

/// Most points an aggregated chart keeps. Past this X is close to a value per row,
/// and a time bucket is what it needs.
pub const AGGREGATE_POINTS_MAX: usize = 200_000;

/// What an aggregated line or scatter chart groups by and makes of the rows.
#[derive(Clone, Copy, Debug)]
pub struct AggregateSpec<'a> {
    pub x: &'a str,
    pub time_unit: crate::chart_modal::TimeUnit,
    pub ys: &'a [String],
    pub aggregate: crate::chart_modal::Aggregate,
    /// The percentile a quantile takes.
    pub quantile: u8,
    pub cumulative: crate::chart_modal::Cumulative,
    pub color: Option<ColorSplit<'a>>,
}

/// Y as an aggregate reads it: as numbers, or as it is for a distinct count, which
/// counts strings and dates too.
fn y_values(y: Expr, aggregate: crate::chart_modal::Aggregate) -> Expr {
    if aggregate.takes_any_y() {
        y
    } else {
        y.cast(DataType::Float64)
    }
}

/// The row index first and last read the rows' order by.
const ROW_ORDER: &str = "__i";

/// `values`' aggregate in a plan. A quantile takes `quantile` percent; first and
/// last go by [`ROW_ORDER`], which the plan must carry, so a group's rows keep the
/// view's order whatever order the engine hands them over in.
fn aggregate_expr(values: Expr, aggregate: crate::chart_modal::Aggregate, quantile: u8) -> Expr {
    use crate::chart_modal::Aggregate;
    let in_order = || {
        values
            .clone()
            .sort_by([col(ROW_ORDER)], SortMultipleOptions::default())
            .drop_nulls()
    };
    match aggregate {
        // Nulls are no value: a group of only nulls has none, a gap.
        Aggregate::Distinct => values.drop_nulls().n_unique().cast(DataType::Float64),
        Aggregate::Sum => values.sum(),
        Aggregate::Mean => values.mean(),
        Aggregate::Median => values.median(),
        // The sample deviation: null for a group of one, which draws no point.
        Aggregate::Stdev => values.std(1),
        Aggregate::Quantile => {
            values.quantile(lit(f64::from(quantile) / 100.0), QuantileMethod::Linear)
        }
        Aggregate::Min => values.min(),
        Aggregate::Max => values.max(),
        Aggregate::First => in_order().first(),
        Aggregate::Last => in_order().last(),
        Aggregate::None | Aggregate::Count => len().cast(DataType::Float64),
    }
}

/// `lf` with [`ROW_ORDER`] when `aggregate` reads the rows' order.
fn with_row_order(lf: &LazyFrame, aggregate: crate::chart_modal::Aggregate) -> LazyFrame {
    if aggregate.follows_row_order() {
        lf.clone().with_row_index(ROW_ORDER, None)
    } else {
        lf.clone()
    }
}

/// Collect an aggregate's plan, streamed whatever the setting: a group-by holds a
/// row per group, and the streaming engine checks `cancel` between morsels. A plan
/// the streaming engine cannot take runs in memory, where the check runs once and
/// the pass goes to its end. A pass stopped by `cancel` is an error that says so.
fn aggregate_pass(lf: LazyFrame, sampling: &ChartSampling) -> Result<DataFrame> {
    crate::statistics::collect_lazy(lf, true).map_err(|e| {
        if sampling.cancel.load(Ordering::Relaxed) {
            color_eyre::eyre::eyre!(ENVELOPE_CANCELLED)
        } else {
            e.into()
        }
    })
}

/// Rows of X a sample reads to judge how many values it has.
const GROUPS_SAMPLE: usize = 20_000;

/// Refuse, before the group-by, an X with more values than a chart can draw: a
/// group per value of a column of nearly as many values as rows would hold the
/// table. Judged from a sample of X: its distinct share, times the rows.
fn refuse_too_many_groups(
    lf: &LazyFrame,
    x: &str,
    most: usize,
    sampling: &ChartSampling,
) -> Result<()> {
    let read = crate::statistics::analysis_rows(
        &lf.clone().select([col(x)]),
        Some(GROUPS_SAMPLE),
        sampling.known_total,
        sampling.seed,
        sampling.streaming,
    )?;
    let distinct = read.df.column(x)?.n_unique()?;
    let read_rows = read.df.height().max(1);
    let estimate = match read.sample_size {
        Some(_) => distinct as f64 / read_rows as f64 * read.total_rows as f64,
        None => distinct as f64,
    };
    if estimate > most as f64 {
        return Err(color_eyre::eyre::eyre!(
            "about {} values of {x}: more than a chart can draw. Bucket X by a time \
             unit, or choose a column with fewer values",
            crate::numfmt::group_chrome(estimate as usize)
        ));
    }
    Ok(())
}

/// A line or scatter chart of Y aggregated per X (per time bucket of a temporal X),
/// and per color group: one lazy group-by over every row of the view. A count needs
/// no Y column; any other aggregate draws a series per Y column, or per color group
/// of the first.
///
/// With cumulative on, each point is the running total of the rows up to the end
/// of its X (its bucket), per series, in X order: a running sum of Y, or Y's rates
/// compounded, `(1 + y1)(1 + y2)... - 1` over every row. Each bucket carries its
/// rows' sum (or the sum of `ln(1 + y)`, which compounds the same), and the totals
/// run across buckets; the aggregate is not used, but a count runs as a count of
/// rows.
pub fn prepare_aggregate_xy(
    lf: &LazyFrame,
    schema: &Schema,
    spec: &AggregateSpec<'_>,
    sampling: &ChartSampling,
) -> Result<GroupedSeries> {
    use crate::chart_modal::{Aggregate, Cumulative};
    let x_dtype = schema
        .get(spec.x)
        .ok_or_else(|| color_eyre::eyre::eyre!("x column '{}' not in schema", spec.x))?;
    let mut x = col(spec.x);
    let bucketed = spec.time_unit.every().is_some()
        && matches!(x_dtype, DataType::Date | DataType::Datetime(_, _));
    if let Some(every) = spec.time_unit.every().filter(|_| bucketed) {
        x = x.dt().truncate(lit(every));
    }
    if !bucketed {
        refuse_too_many_groups(lf, spec.x, AGGREGATE_POINTS_MAX, sampling)?;
    }
    let x = until_cancelled(x, &sampling.cancel).alias("__x");
    let count = spec.aggregate == Aggregate::Count;
    let ys: &[String] = match (count, spec.color) {
        (true, _) => &[],
        (false, Some(_)) => &spec.ys[..spec.ys.len().min(1)],
        (false, None) => spec.ys,
    };
    let mut select = vec![x];
    let mut keys = vec![col("__x")];
    for (i, y) in ys.iter().enumerate() {
        select.push(y_values(col(y.as_str()), spec.aggregate).alias(format!("__y{i}")));
    }
    let plan = with_row_order(lf, spec.aggregate);
    if spec.aggregate.follows_row_order() {
        select.push(col(ROW_ORDER));
    }
    if let Some(color) = spec.color {
        select.push(group_expr(color).alias("__g"));
        keys.push(col("__g"));
    }
    let mut plan = plan.select(select).filter(col("__x").is_not_null());
    if spec.color.is_some() {
        plan = plan.filter(col("__g").is_not_null());
    }
    let mut aggs = vec![len().alias("__n")];
    for i in 0..ys.len() {
        let y = col(format!("__y{i}"));
        let made = match spec.cumulative {
            Cumulative::Off => aggregate_expr(y.clone(), spec.aggregate, spec.quantile),
            Cumulative::Sum => y.clone().sum(),
            Cumulative::Compound => (lit(1.0) + y.clone()).log(lit(std::f64::consts::E)).sum(),
        };
        aggs.push(made.alias(format!("__a{i}")));
        // The values behind it: none is a gap, not the zero a sum of nothing is.
        aggs.push(y.count().alias(format!("__c{i}")));
    }
    let df = aggregate_pass(
        plan.group_by_stable(keys)
            .agg(aggs)
            .sort(["__x"], Default::default()),
        sampling,
    )?;
    if df.height() > AGGREGATE_POINTS_MAX {
        return Err(color_eyre::eyre::eyre!(
            "{} points: more than a chart can draw. Bucket X by a time unit, or \
             choose an X with fewer values",
            crate::numfmt::group_chrome(df.height())
        ));
    }
    let xs: Vec<Option<f64>> = x_values(&df, "__x", x_dtype)?;
    let counts: Vec<u64> = df
        .column("__n")?
        .cast(&DataType::UInt64)?
        .u64()?
        .iter()
        .map(|n| n.unwrap_or(0))
        .collect();
    let groups: Option<Vec<Option<u32>>> = match spec.color {
        Some(_) => Some(df.column("__g")?.u32()?.iter().collect()),
        None => None,
    };
    let values: Vec<Vec<Option<f64>>> = if count {
        vec![counts.iter().map(|&n| Some(n as f64)).collect()]
    } else {
        (0..ys.len())
            .map(|i| {
                let made = df.column(&format!("__a{i}"))?.f64()?.clone();
                let behind = df.column(&format!("__c{i}"))?.cast(&DataType::UInt64)?;
                let behind = behind.u64()?;
                Ok(made
                    .iter()
                    .zip(behind.iter())
                    .map(|(v, n)| {
                        let v = v.filter(|_| n.unwrap_or(0) > 0)?;
                        // A bucket's compound return, from its log sum.
                        Some(if spec.cumulative == Cumulative::Compound {
                            v.exp_m1()
                        } else {
                            v
                        })
                    })
                    .collect())
            })
            .collect::<Result<_>>()?
    };
    let names: Vec<String> = match spec.color {
        Some(color) => color.names(),
        None if count => vec!["count".to_string()],
        None => ys.to_vec(),
    };
    let n = names.len();
    let mut series = vec![Vec::new(); n];
    let mut breaks = vec![Vec::new(); n];
    let mut gap = vec![false; n];
    let mut push = |s: usize, x: f64, y: Option<f64>| match y.filter(|y| y.is_finite()) {
        Some(y) => {
            if gap[s] && !series[s].is_empty() {
                breaks[s].push(series[s].len());
            }
            gap[s] = false;
            series[s].push((x, y));
        }
        None => gap[s] = true,
    };
    for (row, x) in xs.iter().enumerate() {
        let Some(x) = *x else { continue };
        match &groups {
            Some(groups) => {
                if let Some(g) = groups[row] {
                    push(g as usize, x, values[0][row]);
                }
            }
            None => {
                for (s, column) in values.iter().enumerate() {
                    push(s, x, column[row]);
                }
            }
        }
    }
    // A count of rows runs as a count, whichever way the totals were asked to run.
    let how = match spec.cumulative {
        Cumulative::Compound if count => Cumulative::Sum,
        how => how,
    };
    for points in &mut series {
        accumulate(points, how);
    }
    Ok(GroupedSeries {
        names,
        series,
        breaks,
        x_axis_kind: x_axis_temporal_kind(x_dtype),
        rows: RowsRead {
            total_rows: counts.iter().sum::<u64>() as usize,
            sample_size: None,
            envelope_steps: None,
            seed: None,
        },
        other: spec.color.is_some_and(|c| c.other),
    })
}

/// Make `points` cumulative along X: a running sum, or returns compounded (each
/// value a rate; the point is what 1 grew to, less 1).
pub fn accumulate(points: &mut [(f64, f64)], how: crate::chart_modal::Cumulative) {
    use crate::chart_modal::Cumulative;
    let mut total = 0.0;
    for (_, y) in points.iter_mut() {
        total = match how {
            Cumulative::Off => return,
            Cumulative::Sum => total + *y,
            Cumulative::Compound => (1.0 + total) * (1.0 + *y) - 1.0,
        };
        *y = total;
    }
}

/// What an aggregated bar chart groups by and makes of the rows.
#[derive(Clone, Copy, Debug)]
pub struct BarAggregate<'a> {
    pub category: &'a str,
    /// The Y column; none for a count.
    pub value: Option<&'a str>,
    pub aggregate: crate::chart_modal::Aggregate,
    /// The percentile a quantile takes.
    pub quantile: u8,
    pub color: Option<ColorSplit<'a>>,
    pub order: BarOrder,
    pub cap: usize,
}

/// A bar chart of `value` aggregated per category (and per color group): one lazy
/// group-by over every row of the view. A count needs no value column; one without
/// a color is the exact count a bar chart of counts draws.
pub fn prepare_bar_aggregate(
    lf: &LazyFrame,
    spec: &BarAggregate<'_>,
    sampling: &ChartSampling,
) -> Result<BarData> {
    use crate::chart_modal::Aggregate;
    let BarAggregate {
        category,
        value,
        aggregate,
        quantile,
        color,
        order,
        cap,
    } = *spec;
    let count = aggregate == Aggregate::Count;
    if count && color.is_none() {
        return prepare_bar_counts(lf, category, order, cap, sampling);
    }
    let value = match value {
        Some(value) if !count => Some(value),
        None if !count => return Err(color_eyre::eyre::eyre!("Pick a Y column")),
        _ => None,
    };
    let schema = lf.clone().collect_schema()?;
    let value_dtype = match value {
        Some(v) => schema
            .get(v)
            .cloned()
            .ok_or_else(|| color_eyre::eyre::eyre!("no column {v}"))?,
        None => DataType::UInt64,
    };
    let mut select = vec![until_cancelled(col(category), &sampling.cancel)];
    let mut keys = vec![col(category)];
    if let Some(value) = value {
        select.push(y_values(col(value), aggregate).alias("__v"));
    }
    if aggregate.follows_row_order() {
        select.push(col(ROW_ORDER));
    }
    if let Some(color) = color {
        select.push(group_expr(color).alias("__g"));
        keys.push(col("__g"));
    }
    let mut plan = with_row_order(lf, aggregate).select(select);
    if color.is_some() {
        plan = plan.filter(col("__g").is_not_null());
    }
    let measure = match value {
        Some(_) => aggregate_expr(col("__v"), aggregate, quantile),
        None => len().cast(DataType::Float64),
    };
    // The values behind each bar: none is no bar, not the zero a sum of nothing is.
    let behind = match value {
        Some(_) => col("__v").count(),
        None => len(),
    };
    refuse_too_many_groups(lf, category, COUNT_CATEGORY_CAP, sampling)?;
    let df = aggregate_pass(
        // Stable, so bars of equal value keep one order from run to run.
        plan.group_by_stable(keys).agg([
            len().alias("__n"),
            measure.alias("__a"),
            behind.alias("__c"),
        ]),
        sampling,
    )?;
    let rows: usize = df
        .column("__n")?
        .cast(&DataType::UInt64)?
        .u64()?
        .iter()
        .map(|n| n.unwrap_or(0) as usize)
        .sum();
    // A sum, least or greatest of whole numbers is whole; a count always is.
    let whole = aggregate.is_count()
        || (value_dtype.is_integer()
            && matches!(
                aggregate,
                Aggregate::Sum
                    | Aggregate::Min
                    | Aggregate::Max
                    | Aggregate::First
                    | Aggregate::Last
            ));
    let categories = df.column(category)?.as_materialized_series().clone();
    let labels_series = crate::past_calendar::cast_text(&categories, CastOptions::NonStrict)?;
    let labels: Vec<Option<&str>> = labels_series.str()?.iter().collect();
    let behind: Vec<u64> = df
        .column("__c")?
        .cast(&DataType::UInt64)?
        .u64()?
        .iter()
        .map(|n| n.unwrap_or(0))
        .collect();
    // NaN and infinities draw nothing true: no bar.
    let measures: Vec<Option<f64>> = df
        .column("__a")?
        .f64()?
        .iter()
        .zip(&behind)
        .map(|(v, &n)| v.filter(|v| v.is_finite() && n > 0))
        .collect();
    let value_column = match value {
        Some(value) => format!("{} {value}", aggregate.named(quantile)),
        None => "count".to_string(),
    };
    let mut data = BarData {
        category: category.to_string(),
        value_column,
        bars: Vec::new(),
        more: 0,
        no_value: 0,
        rows: RowsRead {
            total_rows: rows,
            sample_size: None,
            envelope_steps: None,
            seed: None,
        },
        value_dtype: if whole {
            DataType::Int64
        } else {
            DataType::Float64
        },
        counted: None,
        groups: Vec::new(),
        other: false,
        rows_note: None,
    };
    let too_many = || {
        color_eyre::eyre::eyre!(
            "more than {} categories of {category}: choose a column with fewer",
            crate::numfmt::group_chrome(COUNT_CATEGORY_CAP)
        )
    };
    let Some(color) = color else {
        if df.height() > COUNT_CATEGORY_CAP {
            return Err(too_many());
        }
        let (bars, more, no_value) = order_bars(&categories, &labels, &measures, order, cap);
        (data.bars, data.more, data.no_value) = (bars, more, no_value);
        return Ok(data);
    };
    // One bar row per category, a value per group: the category's first row stands
    // for it when ordering by label.
    let groups: Vec<Option<u32>> = df.column("__g")?.u32()?.iter().collect();
    let mut at: std::collections::HashMap<Option<&str>, usize> = Default::default();
    let mut firsts: Vec<IdxSize> = Vec::new();
    let mut rows_of: Vec<Vec<Option<f64>>> = Vec::new();
    for (row, label) in labels.iter().enumerate() {
        let i = *at.entry(*label).or_insert_with(|| {
            firsts.push(row as IdxSize);
            rows_of.push(vec![None; color.series()]);
            rows_of.len() - 1
        });
        if let Some(g) = groups[row] {
            rows_of[i][g as usize] = measures[row];
        }
    }
    if rows_of.len() > COUNT_CATEGORY_CAP {
        return Err(too_many());
    }
    let unique = categories.take(&IdxCa::from_vec("firsts".into(), firsts.clone()))?;
    let unique_labels: Vec<Option<&str>> = firsts.iter().map(|&r| labels[r as usize]).collect();
    // A count or a sum adds up across groups; any other measure orders by its
    // largest group.
    let totals: Vec<Option<f64>> = rows_of
        .iter()
        .map(|values| {
            let present = values.iter().flatten();
            if count || aggregate == Aggregate::Sum {
                Some(present.sum())
            } else {
                present.copied().reduce(f64::max)
            }
        })
        .collect();
    let (mut bars, more, no_value) = order_bars(&unique, &unique_labels, &totals, order, cap);
    // `order_bars` names each bar by its label; give each its groups back.
    let by_label: std::collections::HashMap<Option<&str>, usize> = unique_labels
        .iter()
        .enumerate()
        .map(|(i, l)| (*l, i))
        .collect();
    for bar in &mut bars {
        if let Some(&i) = by_label.get(&bar.label.as_deref()) {
            bar.by_group = rows_of[i].clone();
        }
    }
    data.bars = bars;
    data.more = more;
    data.no_value = no_value;
    data.groups = color.names();
    data.other = color.other;
    Ok(data)
}

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

    /// Every label of an axis at one level: its ticks' labels in order.
    fn tick_labels(ticks: &[f64], numbers: &AxisNumbers, level: usize) -> Vec<String> {
        let format = AxisFormat::new(ticks, numbers);
        ticks
            .iter()
            .map(|&v| format.label(v, level).unwrap())
            .collect()
    }

    fn preset(name: &str) -> AxisNumbers {
        AxisNumbers {
            format: crate::numfmt::NumberFormat::preset(name).unwrap(),
            whole: false,
        }
    }

    /// A log axis writes every tick exactly in one form, whatever the closest two
    /// are; its short form names each tick's own k, M or G.
    #[test]
    fn log_axis_labels_write_each_tick_exactly() {
        let ticks = [0.0, 1.0, 10.0, 100.0, 1e3, 1e4, 2e5, 1e6, 5e9];
        let labels = |numbers: &AxisNumbers, level| {
            let format = AxisFormat::log(&ticks, numbers);
            ticks
                .iter()
                .map(|&v| format.label(v, level).unwrap())
                .collect::<Vec<_>>()
        };
        assert_eq!(
            labels(&preset("thousands"), 0),
            [
                "0",
                "1",
                "10",
                "100",
                "1,000",
                "10,000",
                "200,000",
                "1,000,000",
                "5,000,000,000"
            ]
        );
        assert_eq!(
            labels(&AxisNumbers::default(), 1),
            ["0", "1", "10", "100", "1k", "10k", "200k", "1M", "5G"]
        );
        let short = [0.0, 0.25, 0.5, 0.75, 1.0];
        let format = AxisFormat::log(&short, &AxisNumbers::default());
        let written: Vec<_> = short.iter().map(|&v| format.label(v, 0).unwrap()).collect();
        assert_eq!(written, ["0.00", "0.25", "0.50", "0.75", "1.00"]);
        assert_eq!(
            format.label(1.0, 1),
            None,
            "no shorter form under a thousand"
        );
        let huge = AxisFormat::log(&[1e15, 2e16, 1e18], &AxisNumbers::default());
        assert_eq!(huge.label(2e16, 0).as_deref(), Some("2e16"));
    }

    /// Each form a narrow axis steps down to, the same for every tick on it.
    #[test]
    fn axis_labels_step_down_to_shorter_forms() {
        let plain = AxisNumbers::default();
        let ticks = [0.0, 12_345.0, 24_690.0];
        assert_eq!(tick_labels(&ticks, &plain, 0), ["0", "12345", "24690"]);
        assert_eq!(tick_labels(&ticks, &plain, 1), ["0", "12k", "25k"]);
        let ticks = [-1500.0, 0.0, 1500.0];
        assert_eq!(tick_labels(&ticks, &plain, 1), ["-1.5k", "0", "1.5k"]);
        let ticks = [0.0, 1.5e9, 3e9];
        assert_eq!(tick_labels(&ticks, &plain, 1), ["0", "1.5G", "3.0G"]);
        // Below a thousand there is no shorter form.
        let format = AxisFormat::new(&[0.0, 5.0], &plain);
        assert_eq!(format.label(5.0, 1), None);
        assert_eq!(format.label(5.0, 2), None);

        // 2020-01-01 and 2024-12-31 in days; the same instants in microseconds.
        let (lo, hi) = (18262.0, 20088.0);
        let numbers = AxisFormat::new(&[], &plain);
        let date =
            |v, bounds, level| x_axis_label_at(v, XAxisTemporalKind::Date, bounds, level, &numbers);
        let forms: Vec<_> = (0..).map_while(|level| date(hi, (lo, hi), level)).collect();
        assert_eq!(forms, ["2024-12-31", "2024-12", "2024"]);
        // Inside one year, month and day tell the ticks apart.
        assert_eq!(date(lo, (lo, lo + 30.0), 1).as_deref(), Some("01-01"));

        let us = 86_400.0 * 1e6;
        let kind = XAxisTemporalKind::DatetimeUs;
        let at = |v, bounds, level| x_axis_label_at(v, kind, bounds, level, &numbers);
        let forms: Vec<_> = (0..)
            .map_while(|level| at(lo * us, (lo * us, hi * us), level))
            .collect();
        assert_eq!(forms, ["2020-01-01 00:00", "2020-01-01", "2020-01", "2020"]);
        // Inside one day, the time of day.
        let day = (lo * us, lo * us + 3600e6);
        assert_eq!(at(lo * us + 3600e6, day, 1).as_deref(), Some("01:00"));
    }

    /// One format for every label on an axis: densities around 0.01 keep one
    /// precision, where per tick they switched to scientific notation partway up.
    #[test]
    fn an_axis_keeps_one_notation_and_precision() {
        let plain = AxisNumbers::default();
        let density = tick_labels(&[0.0, 0.00651, 0.01302], &plain, 0);
        assert_eq!(density, ["0.0000", "0.0065", "0.0130"]);
        let density = tick_labels(&[0.0, 0.00451, 0.00902], &plain, 0);
        assert_eq!(density, ["0.00000", "0.00451", "0.00902"]);
        // Round ticks take the fewest places that write them all: `0.006`, not
        // `0.0060`; `20`, not `20.0`.
        let round = tick_labels(&[0.0, 0.006, 0.012], &plain, 0);
        assert_eq!(round, ["0.000", "0.006", "0.012"]);
        let round = tick_labels(&[0.0, 0.1 * 3.0, 0.6], &plain, 0);
        assert_eq!(round, ["0.0", "0.3", "0.6"]);
        let round = tick_labels(&[0.0, 20.0, 40.0, 60.0, 80.0], &plain, 0);
        assert_eq!(round, ["0", "20", "40", "60", "80"]);
        let round = tick_labels(&[0.0, 2_000.0, 4_000.0], &plain, 1);
        assert_eq!(round, ["0", "2k", "4k"]);
        // Too small to write in places: scientific, all of them.
        let tiny = tick_labels(&[0.0, 2.5e-8, 5e-8], &plain, 0);
        assert_eq!(tiny, ["0.00e0", "2.50e-8", "5.00e-8"]);
        // Three figures of the largest, and places enough to tell ticks apart.
        assert_eq!(
            tick_labels(&[3.21, 50.17, 97.2], &plain, 0),
            ["3.2", "50.2", "97.2"]
        );
        let close = tick_labels(&[1000.1, 1000.2, 1000.3], &plain, 0);
        assert_eq!(close, ["1000.1", "1000.2", "1000.3"]);
        // Nothing reads as a negative zero, not even a tie formatting rounds to it.
        assert_eq!(tick_labels(&[-0.0001, 1.0], &plain, 0), ["0.00", "1.00"]);
        let padded = [-0.5, 249.75, 500.0];
        assert_eq!(tick_labels(&padded, &plain, 0), ["0", "250", "500"]);
        let padded = [-500.0, 24_750.0, 50_000.0];
        assert_eq!(tick_labels(&padded, &plain, 1), ["0", "25k", "50k"]);
        assert_eq!(tick_labels(&[-0.0, 5e-8], &plain, 0), ["0.00e0", "5.00e-8"]);
        // A tick stepped a hair off zero is zero.
        let stepped = [-2e-8, 1.3e-24, 2e-8];
        assert_eq!(
            tick_labels(&stepped, &plain, 0),
            ["-2.00e-8", "0.00e0", "2.00e-8"]
        );
        // Ticks too close for places, or for three figures of scientific notation,
        // take the figures that tell them apart.
        let close = tick_labels(&[1.0, 1.000_000_1], &plain, 0);
        assert_eq!(close, ["1.0000000e0", "1.0000001e0"]);
        let nanoseconds = [1.727e18, 1.727_05e18, 1.7271e18];
        let labels = tick_labels(&nanoseconds, &plain, 0);
        assert_eq!(labels, ["1.72700e18", "1.72705e18", "1.72710e18"]);
        assert_eq!(tick_labels(&nanoseconds, &plain, 1), labels);
        // Past what places can hold, scientific, and its short form too.
        let huge = [0.0, 5e15];
        assert_eq!(tick_labels(&huge, &plain, 0), ["0.00e0", "5.00e15"]);
        assert_eq!(tick_labels(&huge, &plain, 1), ["0e0", "5e15"]);
        let huge = [1e15, 1.5e15, 2e15];
        assert_eq!(
            tick_labels(&huge, &plain, 1),
            ["1.0e15", "1.5e15", "2.0e15"]
        );
        // A whole-number axis prints whole numbers.
        let whole = AxisNumbers {
            whole: true,
            ..preset("thousands")
        };
        assert_eq!(
            tick_labels(&[0.0, 2161.0, 4322.0], &whole, 0),
            ["0", "2,161", "4,322"]
        );
    }

    /// The table's grouping and decimal separator, in the full form and the short:
    /// `12,3k` under the european format.
    #[test]
    fn axis_labels_take_the_table_number_style() {
        let european = preset("european");
        let ticks = [12_000.0, 12_300.0, 12_600.0];
        assert_eq!(
            tick_labels(&ticks, &european, 0),
            ["12.000", "12.300", "12.600"]
        );
        assert_eq!(
            tick_labels(&ticks, &european, 1),
            ["12,0k", "12,3k", "12,6k"]
        );
        let ticks = [0.0, 0.25, 0.5];
        assert_eq!(tick_labels(&ticks, &european, 0), ["0,00", "0,25", "0,50"]);
        assert_eq!(
            tick_labels(&[0.0, 5e-8], &european, 0),
            ["0,00e0", "5,00e-8"]
        );

        let thousands = preset("thousands");
        let ticks = [0.0, 6172.4, 12345.0];
        assert_eq!(tick_labels(&ticks, &thousands, 0), ["0", "6,172", "12,345"]);
        let ticks = [0.0, 12_300.0];
        assert_eq!(tick_labels(&ticks, &thousands, 1), ["0", "12k"]);
    }

    fn all_rows() -> ChartSampling {
        ChartSampling::rows(Some(10_000))
    }

    fn xy(lf: &LazyFrame, x: &str, ys: &[&str], sampling: &ChartSampling) -> ChartDataResult {
        let schema = lf.clone().collect_schema().unwrap();
        let ys: Vec<String> = ys.iter().map(|s| s.to_string()).collect();
        prepare_chart_data(lf, schema.as_ref(), x, &ys, sampling, false).unwrap()
    }

    /// A line over more rows than its sample size draws each step's lowest and
    /// highest value: every peak survives, where a sample of a waveform misses them.
    #[test]
    fn a_long_line_is_drawn_as_its_envelope() {
        let n = 100_000usize;
        let x: Vec<i64> = (0..n as i64).collect();
        let y: Vec<f64> = (0..n)
            .map(|i| match i {
                // One spike, one row wide, that a sample would almost surely miss.
                54_321 => 9.0,
                _ => ((i as f64) / 50.0).sin(),
            })
            .collect();
        let lf = df!("x" => &x, "y" => &y).unwrap().lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let sampling = ChartSampling::rows(Some(1_000));
        let result =
            prepare_chart_data(&lf, schema.as_ref(), "x", &["y".into()], &sampling, true).unwrap();
        assert_eq!(
            result.rows,
            RowsRead {
                total_rows: n,
                sample_size: None,
                envelope_steps: Some(500),
                seed: None,
            }
        );
        let points = &result.series[0];
        assert!(points.len() <= 1_000, "{} points", points.len());
        let top = points.iter().map(|p| p.1).fold(f64::MIN, f64::max);
        assert_eq!(top, 9.0, "the spike is kept");
        let bottom = points.iter().map(|p| p.1).fold(f64::MAX, f64::min);
        assert!(bottom < -0.99, "so is every trough: {bottom}");
        assert!(points.windows(2).all(|w| w[0].0 <= w[1].0), "in X order");
        assert_eq!(
            chart_notes(&result.rows, None, "·"),
            ["min and max of 100k rows in 500 steps"]
        );

        // Under the sample size, every row is drawn as it was.
        let sampling = ChartSampling::rows(Some(200_000));
        let result =
            prepare_chart_data(&lf, schema.as_ref(), "x", &["y".into()], &sampling, true).unwrap();
        assert_eq!(result.rows.envelope_steps, None);
        assert_eq!(result.series[0].len(), n);
    }

    /// An envelope reads the whole view twice: never over an object store in place,
    /// where the sample reads a few row groups; and both passes stop when the chart
    /// is no longer wanted.
    #[test]
    fn an_envelope_is_sampled_instead_where_full_reads_cost_and_stops_when_cancelled() {
        let n = 10_000i64;
        let lf = df!("x" => (0..n).collect::<Vec<_>>(), "y" => (0..n).collect::<Vec<_>>())
            .unwrap()
            .lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let remote = ChartSampling {
            full_passes: false,
            ..ChartSampling::rows(Some(100))
        };
        let result =
            prepare_chart_data(&lf, schema.as_ref(), "x", &["y".into()], &remote, true).unwrap();
        assert_eq!(result.rows.envelope_steps, None);
        assert_eq!(result.rows.sample_size, Some(100));

        let cancelled = ChartSampling::rows(Some(100));
        cancelled.cancel.store(true, Ordering::Relaxed);
        let err = prepare_chart_data(&lf, schema.as_ref(), "x", &["y".into()], &cancelled, true)
            .err()
            .expect("a cancelled envelope is not drawn");
        assert_eq!(err.to_string(), "chart cancelled");
    }

    /// A temporal X is placed by its ordinal, as a sampled line places it.
    #[test]
    fn an_envelope_places_temporal_x_by_its_ordinal() {
        let days: Vec<i32> = (0..1_000).collect();
        let lf = df!("d" => &days, "y" => (0..1_000).map(f64::from).collect::<Vec<_>>())
            .unwrap()
            .lazy()
            .with_column(col("d").cast(DataType::Date));
        let schema = lf.clone().collect_schema().unwrap();
        let result = prepare_chart_data(
            &lf,
            schema.as_ref(),
            "d",
            &["y".into()],
            &ChartSampling::rows(Some(100)),
            true,
        )
        .unwrap();
        assert_eq!(result.rows.envelope_steps, Some(50));
        let points = &result.series[0];
        assert_eq!(points.first(), Some(&(0.0, 0.0)));
        assert_eq!(points.last().map(|p| p.1), Some(999.0));
        assert!(
            points.iter().all(|&(x, y)| y >= x && y < x + 20.0),
            "{points:?}"
        );
    }

    /// A step where a series has no value breaks its line, as a null does.
    #[test]
    fn an_envelope_breaks_where_a_series_has_no_values() {
        let x: Vec<i64> = (0..100).collect();
        let y: Vec<Option<f64>> = (0..100)
            .map(|i| (!(40..60).contains(&i)).then_some(i as f64))
            .collect();
        let lf = df!("x" => &x, "y" => &y).unwrap().lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let sampling = ChartSampling::rows(Some(20));
        let result =
            prepare_chart_data(&lf, schema.as_ref(), "x", &["y".into()], &sampling, true).unwrap();
        assert_eq!(result.rows.envelope_steps, Some(10));
        assert_eq!(result.breaks[0].len(), 1, "one gap: {:?}", result.series[0]);
    }

    /// A row whose X is not a number is left out whole, Y and all.
    #[test]
    fn an_envelope_leaves_out_rows_with_no_x() {
        let x: Vec<f64> = (0..100)
            .map(|i| if i % 10 == 0 { f64::NAN } else { i as f64 })
            .collect();
        let y: Vec<f64> = (0..100).map(|i| i as f64).collect();
        let lf = df!("x" => &x, "y" => &y).unwrap().lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let sampling = ChartSampling::rows(Some(20));
        let result =
            prepare_chart_data(&lf, schema.as_ref(), "x", &["y".into()], &sampling, true).unwrap();
        let ys: Vec<f64> = result.series[0].iter().map(|p| p.1).collect();
        assert!(!ys.contains(&0.0) && !ys.contains(&50.0), "{ys:?}");
        assert_eq!(ys.iter().cloned().fold(f64::MIN, f64::max), 99.0);
    }

    #[test]
    fn prepare_empty_y_columns() {
        let lf = df!("x" => &[1.0_f64, 2.0], "y" => &[10.0, 20.0])
            .unwrap()
            .lazy();
        let result = xy(&lf, "x", &[], &all_rows());
        assert!(result.series.is_empty());
        assert_eq!(result.x_axis_kind, XAxisTemporalKind::Numeric);
    }

    #[test]
    fn prepare_small_data() {
        let lf = df!(
            "x" => &[1.0_f64, 2.0, 3.0],
            "a" => &[10.0_f64, 20.0, 30.0],
            "b" => &[100.0_f64, 200.0, 300.0]
        )
        .unwrap()
        .lazy();
        let result = xy(&lf, "x", &["a", "b"], &all_rows());
        assert_eq!(result.series.len(), 2);
        assert_eq!(
            result.series[0],
            vec![(1.0, 10.0), (2.0, 20.0), (3.0, 30.0)]
        );
        assert_eq!(
            result.series[1],
            vec![(1.0, 100.0), (2.0, 200.0), (3.0, 300.0)]
        );
        assert_eq!(result.x_axis_kind, XAxisTemporalKind::Numeric);
        assert_eq!(
            result.rows,
            RowsRead {
                total_rows: 3,
                sample_size: None,
                envelope_steps: None,
                seed: None,
            },
            "every row read: nothing to say"
        );
        assert!(chart_notes(&result.rows, None, "·").is_empty());
    }

    #[test]
    fn prepare_skips_nan() {
        let lf = df!(
            "x" => &[1.0_f64, 2.0, 3.0],
            "y" => &[10.0_f64, f64::NAN, 30.0]
        )
        .unwrap()
        .lazy();
        let result = xy(&lf, "x", &["y"], &all_rows());
        assert_eq!(result.series[0], vec![(1.0, 10.0), (3.0, 30.0)]);
    }

    #[test]
    fn prepare_missing_x_column_errors() {
        let lf = df!("x" => &[1.0_f64], "y" => &[2.0_f64]).unwrap().lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let result = prepare_chart_data(
            &lf,
            schema.as_ref(),
            "missing",
            &["y".into()],
            &all_rows(),
            false,
        );
        assert!(result.is_err());
    }

    /// Over the limit, a chart reads a sample spread across the table, not its head,
    /// and says how many rows it read of how many.
    #[test]
    fn over_the_limit_a_chart_reads_a_spread_sample_and_says_so() {
        let n = 50_000_i64;
        let lf = df!(
            "x" => (0..n).collect::<Vec<_>>(),
            "y" => (0..n).map(|v| v * 2).collect::<Vec<_>>()
        )
        .unwrap()
        .lazy();
        let result = xy(&lf, "x", &["y"], &ChartSampling::rows(Some(1_000)));
        let points = &result.series[0];
        assert_eq!(points.len(), 1_000);
        let last_x = points.last().unwrap().0;
        assert!(
            last_x > (n as f64) * 0.9,
            "the sample reaches the end of the table, got {last_x}"
        );
        assert_eq!(
            result.rows,
            RowsRead {
                total_rows: n as usize,
                sample_size: Some(1_000),
                envelope_steps: None,
                seed: Some(crate::sampling::Sample::default().seed),
            }
        );
        // The seed draws the same sample again; the terminal joins it as ASCII.
        let seed = crate::sampling::Sample::default().seed;
        assert_eq!(
            chart_notes(&result.rows, None, "·"),
            [format!("sample of 1,000 of 50k rows · seed {seed}")]
        );
        assert_eq!(
            chart_notes(&result.rows, None, "-"),
            [format!("sample of 1,000 of 50k rows - seed {seed}")]
        );

        // No limit reads every row.
        let every = xy(&lf, "x", &["y"], &ChartSampling::rows(None));
        assert_eq!(every.series[0].len(), n as usize);
        assert_eq!(every.rows.sample_size, None);
    }

    /// One Parquet file is sampled in runs across it: the chart's columns stay a plan
    /// the sampler can seek in, so the chart does not read the file to draw from it.
    #[test]
    fn a_parquet_file_is_sampled_in_runs() {
        let dir = tempfile::tempdir().unwrap();
        let n = 100_000_i64;
        let mut df = df!(
            "id" => (0..n).collect::<Vec<_>>(),
            "fare" => (0..n).map(|v| v as f64).collect::<Vec<_>>(),
            "other" => vec!["x"; n as usize]
        )
        .unwrap();
        let path = dir.path().join("trips.parquet");
        ParquetWriter::new(std::fs::File::create(&path).unwrap())
            .with_row_group_size(Some(1_000))
            .finish(&mut df)
            .unwrap();
        let lf =
            LazyFrame::scan_parquet(PlRefPath::try_from_path(&path).unwrap(), Default::default())
                .unwrap();
        assert!(crate::statistics::slices_reach_into_the_scan(
            &lf.clone().select([col("id"), col("fare")])
        ));
        let data = prepare_histogram_data(
            &lf,
            "fare",
            10,
            ValueRange::All,
            &ChartSampling::rows(Some(2_000)),
        )
        .unwrap();
        assert_eq!(
            data.rows,
            RowsRead {
                total_rows: n as usize,
                sample_size: Some(2_000),
                envelope_steps: None,
                seed: Some(crate::sampling::Sample::default().seed),
            }
        );
        assert!(data.x_max > 90_000.0, "reaches the end: {}", data.x_max);
    }

    /// The same seed and size draw the same rows; the chart and the analysis tools
    /// share the sampler, so they agree on what a sample is.
    #[test]
    fn the_sample_is_seeded() {
        let lf = df!("x" => (0..20_000_i64).collect::<Vec<_>>(), "y" => vec![1.0_f64; 20_000])
            .unwrap()
            .lazy();
        let a = xy(&lf, "x", &["y"], &ChartSampling::rows(Some(500)));
        let b = xy(&lf, "x", &["y"], &ChartSampling::rows(Some(500)));
        assert_eq!(a.series, b.series);
        let other = ChartSampling {
            seed: 7,
            ..ChartSampling::rows(Some(500))
        };
        let c = xy(&lf, "x", &["y"], &other);
        assert_ne!(a.series, c.series);
    }

    /// Another option over columns already read draws from the rows held, without
    /// reading the file again; a new column is read with the held ones, and another
    /// sample size reads afresh.
    #[test]
    fn rows_already_read_are_not_read_again() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("fares.csv");
        let write = |a: i64, b: i64| {
            let rows: String = (0..100).map(|i| format!("{},{}\n", i + a, i + b)).collect();
            std::fs::write(&path, format!("a,b\n{rows}")).unwrap();
        };
        write(0, 0);
        let lf = LazyCsvReader::new(PlRefPath::try_from_path(&path).unwrap())
            .finish()
            .unwrap();
        let sampling = ChartSampling::rows(Some(10_000));
        let first = prepare_histogram_data(&lf, "a", 10, ValueRange::All, &sampling).unwrap();
        assert_eq!(first.x_min, 0.0);

        write(1_000, 1_000);
        let held =
            prepare_histogram_data(&lf, "a", 5, ValueRange::Percentile1To99, &sampling).unwrap();
        assert!(
            held.x_max < 100.0,
            "drawn from the rows held: {}",
            held.x_max
        );
        let boxed = prepare_box_plot_data(&lf, &["a"], ValueRange::All, &sampling).unwrap();
        assert_eq!(boxed.stats[0].max, 99.0);

        let with_b = prepare_histogram_data(&lf, "b", 10, ValueRange::All, &sampling).unwrap();
        assert_eq!(with_b.x_min, 1_000.0, "b was not held: read");
        let a_again = prepare_histogram_data(&lf, "a", 10, ValueRange::All, &sampling).unwrap();
        assert_eq!(a_again.x_min, 1_000.0, "read along with b");

        write(5_000, 5_000);
        let other_size = ChartSampling {
            limit: Some(50),
            ..sampling.clone()
        };
        let resampled = prepare_histogram_data(&lf, "a", 10, ValueRange::All, &other_size).unwrap();
        assert!(resampled.x_min >= 5_000.0, "another size reads afresh");
    }

    /// A line is drawn in X order whatever order the rows are in, as a pivot leaves
    /// them.
    #[test]
    fn points_come_in_x_order() {
        let lf = df!(
            "year" => &[2001_i64, 1999, 2003, 2000, 2002],
            "count" => &[1.0_f64, 2.0, 3.0, 4.0, 5.0]
        )
        .unwrap()
        .lazy();
        let result = xy(&lf, "year", &["count"], &all_rows());
        let xs: Vec<f64> = result.series[0].iter().map(|p| p.0).collect();
        assert_eq!(xs, [1999.0, 2000.0, 2001.0, 2002.0, 2003.0]);
        assert_eq!(result.series[0][0], (1999.0, 2.0));
        assert!(result.breaks[0].is_empty());
    }

    /// Picking the X column as a Y series charts it against itself rather than failing
    /// on a repeated column.
    #[test]
    fn x_as_a_y_series_charts_rather_than_failing() {
        let lf = df!("x" => &[1.0_f64, 2.0], "y" => &[3.0_f64, 4.0])
            .unwrap()
            .lazy();
        let result = xy(&lf, "x", &["x", "y"], &all_rows());
        assert_eq!(result.series[0], vec![(1.0, 1.0), (2.0, 2.0)]);
        assert_eq!(result.series[1], vec![(1.0, 3.0), (2.0, 4.0)]);
    }

    /// A Y column the frame does not have is an error to show, not an empty chart.
    #[test]
    fn a_missing_y_column_is_an_error() {
        let lf = df!("x" => &[1.0_f64], "y" => &[2.0_f64]).unwrap().lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let result = prepare_chart_data(
            &lf,
            schema.as_ref(),
            "x",
            &["gone".into()],
            &all_rows(),
            false,
        );
        assert!(result.is_err());
    }

    /// One series' nulls drop that series' points only; a null X drops the row; a
    /// line breaks at a gap instead of bridging it.
    #[test]
    fn nulls_drop_per_series_and_break_the_line() {
        let lf = df!(
            "year" => &[Some(1880_i64), Some(1881), Some(1882), None, Some(1883), Some(1884)],
            "emma" => &[Some(10.0_f64), Some(11.0), Some(12.0), Some(99.0), Some(13.0), Some(14.0)],
            "jennifer" => &[None, None, Some(5.0_f64), Some(99.0), None, Some(7.0)]
        )
        .unwrap()
        .lazy();
        let result = xy(&lf, "year", &["emma", "jennifer"], &all_rows());
        assert_eq!(
            result.series[0],
            vec![
                (1880.0, 10.0),
                (1881.0, 11.0),
                (1882.0, 12.0),
                (1883.0, 13.0),
                (1884.0, 14.0)
            ],
            "Emma keeps the years Jennifer is missing; the null year is gone"
        );
        assert!(result.breaks[0].is_empty());
        assert_eq!(result.series[1], vec![(1882.0, 5.0), (1884.0, 7.0)]);
        assert_eq!(result.breaks[1], [1], "1883 is missing: the line breaks");
        assert_eq!(
            segments(&result.series[1], &result.breaks[1]),
            vec![&[(1882.0, 5.0)][..], &[(1884.0, 7.0)][..]]
        );
    }

    #[test]
    fn segments_split_at_breaks() {
        let points = [(0.0, 0.0), (1.0, 1.0), (2.0, 2.0), (3.0, 3.0)];
        assert_eq!(segments(&points, &[]), vec![&points[..]]);
        assert_eq!(
            segments(&points, &[1, 3]),
            vec![&points[..1], &points[1..3], &points[3..]]
        );
        assert!(segments(&[], &[]).is_empty());
    }

    /// Temporal X charts as its ordinal, nulls dropped the same way.
    #[test]
    fn a_date_x_is_ordinal() {
        let lf = df!("d" => &[Some(1_i32), None, Some(0)], "y" => &[1.0_f64, 2.0, 3.0])
            .unwrap()
            .lazy()
            .with_column(col("d").cast(DataType::Date));
        let result = xy(&lf, "d", &["y"], &all_rows());
        assert_eq!(result.x_axis_kind, XAxisTemporalKind::Date);
        assert_eq!(result.series[0], vec![(0.0, 3.0), (1.0, 1.0)]);
    }

    fn with_outliers() -> LazyFrame {
        // 1..=100 and two far outliers.
        let mut v: Vec<f64> = (1..=100).map(f64::from).collect();
        v.push(-10_000.0);
        v.push(50_000.0);
        df!("fare" => v).unwrap().lazy()
    }

    /// The percentile range leaves the tails out of a histogram and counts them.
    #[test]
    fn a_histogram_range_clips_the_tails_and_counts_them() {
        let lf = with_outliers();
        let all = prepare_histogram_data(&lf, "fare", 10, ValueRange::All, &all_rows()).unwrap();
        assert_eq!(all.x_min, -10_000.0);
        assert!(all.clipped.is_none());

        let clipped =
            prepare_histogram_data(&lf, "fare", 10, ValueRange::Percentile1To99, &all_rows())
                .unwrap();
        // Of 102 values the 1st percentile falls at 1.01 and the 99th at 99.99: each
        // tail loses its outlier and the value next to it.
        assert_eq!((clipped.x_min, clipped.x_max), (2.0, 99.0));
        let outside = clipped.clipped.unwrap().outside;
        assert_eq!(outside, 4);
        let counted: f64 = clipped.bins.iter().map(|b| b.count).sum();
        assert_eq!(counted as usize + outside, 102);
        assert_eq!(
            chart_notes(&clipped.rows, clipped.clipped.as_ref(), "·"),
            ["4 values outside p1-p99"]
        );
    }

    #[test]
    fn box_plot_and_kde_take_the_range_too() {
        let lf = with_outliers();
        let boxed = prepare_box_plot_data(&lf, &["fare"], ValueRange::Percentile1To99, &all_rows())
            .unwrap();
        assert!(boxed.stats[0].min > 0.0 && boxed.stats[0].max <= 100.0);
        assert!(boxed.clipped.unwrap().outside >= 2);

        let kde = prepare_kde_data(
            &lf,
            &["fare"],
            1.0,
            ValueRange::Percentile1To99,
            &all_rows(),
        )
        .unwrap();
        assert!(kde.x_min > -1_000.0 && kde.x_max < 1_000.0);
        assert!(kde.clipped.unwrap().outside >= 2);

        let whole = prepare_box_plot_data(&lf, &["fare"], ValueRange::All, &all_rows()).unwrap();
        assert_eq!(whole.stats[0].min, -10_000.0);
    }

    /// A value column read on its own: another column's nulls do not remove its values.
    #[test]
    fn a_histogram_keeps_every_value_of_its_column() {
        let lf = df!("a" => &[Some(1.0_f64), Some(2.0), None, Some(4.0)])
            .unwrap()
            .lazy();
        let data = prepare_histogram_data(&lf, "a", 5, ValueRange::All, &all_rows()).unwrap();
        let counted: f64 = data.bins.iter().map(|b| b.count).sum();
        assert_eq!(counted, 3.0);
    }

    #[test]
    fn prepare_x_range_numeric() {
        let lf = df!("x" => &[10.0_f64, 20.0, 5.0, 30.0]).unwrap().lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let r = prepare_chart_x_range(&lf, schema.as_ref(), "x", &all_rows()).unwrap();
        assert_eq!(r.x_min, 5.0);
        assert_eq!(r.x_max, 30.0);
        assert_eq!(r.x_axis_kind, XAxisTemporalKind::Numeric);
    }

    #[test]
    fn prepare_x_range_empty_returns_placeholder() {
        let lf = df!("x" => &[1.0_f64]).unwrap().lazy().slice(0, 0);
        let schema = lf.clone().collect_schema().unwrap();
        let r = prepare_chart_x_range(&lf, schema.as_ref(), "x", &all_rows()).unwrap();
        assert_eq!(r.x_min, 0.0);
        assert_eq!(r.x_max, 1.0);
    }

    fn bars(lf: &LazyFrame, order: BarOrder, cap: usize) -> BarData {
        prepare_bar_data(lf, "carrier", "delay", order, cap, &all_rows()).unwrap()
    }

    fn labels(data: &BarData) -> Vec<Option<&str>> {
        data.bars.iter().map(|b| b.label.as_deref()).collect()
    }

    /// Bars come largest first, or in the category's own order; past the cap the rest
    /// are counted rather than kept.
    #[test]
    fn bars_order_by_value_or_label_and_cap_the_rest() {
        let lf = df!(
            "carrier" => &["UA", "AA", "DL", "B6", "AS"],
            "delay" => &[3.5_f64, 0.4, 1.6, 9.5, -9.9]
        )
        .unwrap()
        .lazy();
        let by_value = bars(&lf, BarOrder::Value, BAR_CAP);
        assert_eq!(
            labels(&by_value),
            [Some("B6"), Some("UA"), Some("DL"), Some("AA"), Some("AS")]
        );
        assert_eq!(by_value.bars[4].value, -9.9);
        assert_eq!(by_value.more, 0);

        let by_label = bars(&lf, BarOrder::Label, BAR_CAP);
        assert_eq!(
            labels(&by_label),
            [Some("AA"), Some("AS"), Some("B6"), Some("DL"), Some("UA")]
        );

        let capped = bars(&lf, BarOrder::Value, 2);
        assert_eq!(labels(&capped), [Some("B6"), Some("UA")]);
        assert_eq!(capped.more, 3, "the three smallest are counted, not drawn");
        let capped = bars(&lf, BarOrder::Label, 2);
        assert_eq!(labels(&capped), [Some("AA"), Some("AS")]);
        assert_eq!(capped.more, 3);
    }

    /// An integer category orders as numbers, not text; a null category is a bar of its
    /// own, last in label order; a null value leaves its category out and is counted.
    #[test]
    fn bar_categories_keep_their_type_and_nulls_are_counted() {
        let lf = df!(
            "carrier" => &[Some(10_i64), Some(9), None, Some(100), Some(2)],
            "delay" => &[Some(1.0_f64), Some(2.0), Some(3.0), None, Some(1.0)]
        )
        .unwrap()
        .lazy();
        let data = bars(&lf, BarOrder::Label, BAR_CAP);
        assert_eq!(labels(&data), [Some("2"), Some("9"), Some("10"), None]);
        assert_eq!(data.no_value, 1, "100 has no value");

        let data = bars(&lf, BarOrder::Value, BAR_CAP);
        assert_eq!(
            labels(&data),
            [None, Some("9"), Some("10"), Some("2")],
            "ties keep table order"
        );
    }

    /// A date category past the calendar is labeled by its stored number, as the
    /// table shows it, where the cast to text panicked (#506); bars still order as
    /// dates.
    #[test]
    fn a_date_category_past_the_calendar_is_labeled_by_its_stored_number() {
        let paris = TimeZone::opt_try_new(Some("Europe/Paris")).unwrap();
        let datetime = |unit, zone: Option<TimeZone>| {
            Series::new("at".into(), [i64::MIN + 1, 0])
                .cast(&DataType::Datetime(unit, zone))
                .unwrap()
        };
        for (at, labels_in_order) in [
            (
                Series::new("at".into(), [i32::MAX, 0])
                    .cast(&DataType::Date)
                    .unwrap(),
                ["1970-01-01", "2147483647 days since 1970-01-01"],
            ),
            (
                datetime(TimeUnit::Milliseconds, None),
                [
                    "-9223372036854775807 ms since 1970-01-01 UTC",
                    "1970-01-01 00:00:00.000",
                ],
            ),
            (
                datetime(TimeUnit::Microseconds, paris),
                [
                    "-9223372036854775807 us since 1970-01-01 UTC",
                    "1970-01-01 01:00:00.000000+01:00",
                ],
            ),
        ] {
            let lf = DataFrame::new_infer_height(vec![
                at.into_column(),
                Column::new("n".into(), [1i64, 2]),
            ])
            .unwrap()
            .lazy();
            let by_value =
                prepare_bar_data(&lf, "at", "n", BarOrder::Label, BAR_CAP, &all_rows()).unwrap();
            let counted =
                prepare_bar_counts(&lf, "at", BarOrder::Label, BAR_CAP, &all_rows()).unwrap();
            for data in [by_value, counted] {
                assert_eq!(labels(&data), labels_in_order.map(Some));
            }
        }
    }

    /// A category that repeats is refused with the way out, not summed or averaged.
    #[test]
    fn a_repeated_category_is_refused() {
        let lf = df!(
            "species" => &["Adelie", "Adelie", "Gentoo"],
            "body_mass_g" => &[3750_i64, 3800, 5000]
        )
        .unwrap()
        .lazy();
        let err = prepare_bar_data(
            &lf,
            "species",
            "body_mass_g",
            BarOrder::Value,
            BAR_CAP,
            &all_rows(),
        )
        .unwrap_err()
        .to_string();
        assert!(
            err.contains("species repeats: 2 categories in 3 rows"),
            "{err}"
        );
        assert!(
            err.contains("SELECT species, AVG(body_mass_g) FROM df GROUP BY species"),
            "SQL first: {err}"
        );
        assert!(
            err.contains("(or select avg body_mass_g by species)"),
            "{err}"
        );

        // A name SQL cannot read bare is quoted, and the q form, which cannot, is left out.
        let lf = df!("Species" => &["a", "a"], "mass g" => &[1_i64, 2])
            .unwrap()
            .lazy();
        let err = prepare_bar_data(
            &lf,
            "Species",
            "mass g",
            BarOrder::Value,
            BAR_CAP,
            &all_rows(),
        )
        .unwrap_err()
        .to_string();
        assert!(
            err.ends_with(
                r#"SELECT "Species", AVG("mass g") FROM df GROUP BY "Species", or choose Count for the rows per category"#
            ),
            "{err}"
        );
    }

    /// Another order draws from the rows already read: the file is not read again.
    #[test]
    fn a_new_bar_order_does_not_read_again() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("delays.csv");
        std::fs::write(&path, "carrier,delay\nUA,3.5\nAA,0.4\n").unwrap();
        let lf = LazyCsvReader::new(PlRefPath::try_from_path(&path).unwrap())
            .finish()
            .unwrap();
        let sampling = all_rows();
        let first =
            prepare_bar_data(&lf, "carrier", "delay", BarOrder::Value, BAR_CAP, &sampling).unwrap();
        assert_eq!(labels(&first), [Some("UA"), Some("AA")]);
        std::fs::write(&path, "carrier,delay\nZZ,1.0\n").unwrap();
        let again =
            prepare_bar_data(&lf, "carrier", "delay", BarOrder::Label, BAR_CAP, &sampling).unwrap();
        assert_eq!(
            labels(&again),
            [Some("AA"), Some("UA")],
            "from the rows held"
        );
    }

    /// Booleans and categoricals are categories too.
    #[test]
    fn booleans_and_categoricals_chart_as_categories() {
        let lf = df!("flag" => &[true, false], "n" => &[5_i64, 7])
            .unwrap()
            .lazy();
        let data =
            prepare_bar_data(&lf, "flag", "n", BarOrder::Label, BAR_CAP, &all_rows()).unwrap();
        assert_eq!(labels(&data), [Some("false"), Some("true")]);

        let lf = df!("kind" => &["b", "a"], "n" => &[5_i64, 7])
            .unwrap()
            .lazy()
            .with_column(col("kind").cast(DataType::from_categories(Categories::global())));
        let schema = lf.clone().collect_schema().unwrap();
        assert!(is_category_dtype(schema.get("kind").unwrap()));
        let data =
            prepare_bar_data(&lf, "kind", "n", BarOrder::Value, BAR_CAP, &all_rows()).unwrap();
        assert_eq!(labels(&data), [Some("a"), Some("b")]);
        assert!(!is_category_dtype(&DataType::Float64));
        assert!(is_category_dtype(&DataType::UInt8));
    }

    /// Bar values print in the table's number format: an integer column whole, any
    /// other to the format's places or two, the same for every bar.
    #[test]
    fn bar_values_follow_the_table_number_format() {
        use crate::numfmt::NumberFormat;
        let plain = NumberFormat::PLAIN;
        let thousands = NumberFormat::preset("thousands").unwrap();
        let european = NumberFormat::preset("european").unwrap();
        assert_eq!(format_bar_value(1_234_567.0, true, &plain), "1234567");
        assert_eq!(format_bar_value(1_234_567.0, true, &thousands), "1,234,567");
        assert_eq!(format_bar_value(22.0, false, &plain), "22.00");
        assert_eq!(format_bar_value(-9.9296, false, &plain), "-9.93");
        assert_eq!(format_bar_value(4213.7, false, &thousands), "4,213.70");
        assert_eq!(format_bar_value(4213.7, false, &european), "4.213,70");
        let one_place = NumberFormat {
            float_precision: Some(1),
            ..thousands
        };
        assert_eq!(format_bar_value(4213.74, false, &one_place), "4,213.7");
        assert_eq!(format_bar_value(0.0, false, &plain), "0.00");
        assert_eq!(format_bar_value(0.001, false, &plain), "1.00e-3");

        let lf = df!("carrier" => &["UA", "AA"], "delay" => &[1234.5_f64, 7.0])
            .unwrap()
            .lazy();
        let data = bars(&lf, BarOrder::Value, BAR_CAP);
        let mut settings = crate::numfmt::NumberFormatSettings {
            format: NumberFormat::preset("thousands").unwrap(),
            ..Default::default()
        };
        assert_eq!(data.value_labels(&settings), ["1,234.50", "7.00"]);
        settings.enabled = false;
        assert_eq!(
            data.value_labels(&settings),
            ["1234.50", "7.00"],
            "F turns it off"
        );
    }

    fn species(n_adelie: usize, n_gentoo: usize, n_chinstrap: usize, n_null: usize) -> LazyFrame {
        let mut species: Vec<Option<&str>> = Vec::new();
        // Interleaved, so no stretch of the table is one species.
        let mut left = [
            (Some("Adelie"), n_adelie),
            (Some("Gentoo"), n_gentoo),
            (Some("Chinstrap"), n_chinstrap),
            (None, n_null),
        ];
        while left.iter().any(|(_, n)| *n > 0) {
            for (name, n) in &mut left {
                if *n > 0 {
                    species.push(*name);
                    *n -= 1;
                }
            }
        }
        df!("species" => species).unwrap().lazy()
    }

    fn counts(data: &BarData) -> Vec<(Option<&str>, f64)> {
        data.bars
            .iter()
            .map(|b| (b.label.as_deref(), b.value))
            .collect()
    }

    /// Count is exact over the whole view, not a count of the sample: more rows than
    /// the sample size are all counted, a null category is a bar of its own, and the
    /// note says the counts are of every row.
    #[test]
    fn counts_are_exact_past_the_sample_size() {
        let lf = species(30_000, 15_000, 4_999, 1);
        let sampling = ChartSampling::rows(Some(1_000));
        let data = prepare_bar_counts(&lf, "species", BarOrder::Value, BAR_CAP, &sampling).unwrap();
        assert_eq!(
            counts(&data),
            [
                (Some("Adelie"), 30_000.0),
                (Some("Gentoo"), 15_000.0),
                (Some("Chinstrap"), 4_999.0),
                (None, 1.0)
            ]
        );
        assert_eq!(data.counted, Some(50_000), "counted past the sample size");
        assert_eq!(data.rows.sample_size, None);
        assert_eq!(data.value_column, "count");
        assert_eq!(
            data.value_labels(&crate::numfmt::NumberFormatSettings {
                format: crate::numfmt::NumberFormat::preset("thousands").unwrap(),
                ..Default::default()
            }),
            ["30,000", "15,000", "4,999", "1"],
            "whole numbers"
        );

        let by_label =
            prepare_bar_counts(&lf, "species", BarOrder::Label, BAR_CAP, &sampling).unwrap();
        assert_eq!(
            counts(&by_label),
            [
                (Some("Adelie"), 30_000.0),
                (Some("Chinstrap"), 4_999.0),
                (Some("Gentoo"), 15_000.0),
                (None, 1.0)
            ],
            "the null category last"
        );

        // Every row read, or a view under the sample size: nothing to say.
        let every = prepare_bar_counts(
            &lf,
            "species",
            BarOrder::Value,
            BAR_CAP,
            &ChartSampling::rows(None),
        )
        .unwrap();
        assert_eq!(every.counted, None);
        let small = species(152, 124, 68, 0);
        let data =
            prepare_bar_counts(&small, "species", BarOrder::Value, BAR_CAP, &all_rows()).unwrap();
        assert_eq!(
            counts(&data),
            [
                (Some("Adelie"), 152.0),
                (Some("Gentoo"), 124.0),
                (Some("Chinstrap"), 68.0)
            ]
        );
        assert_eq!(data.counted, None);
    }

    /// Equal counts come A to Z; past the bar cap the rest are counted, not drawn; past
    /// the category cap the count stops and says so rather than drawing a part.
    #[test]
    fn counts_cap_their_bars_and_stop_past_the_category_cap() {
        let lf = df!("carrier" => &["UA", "B6", "AA", "AA", "DL", "B6", "AA"])
            .unwrap()
            .lazy();
        let data = prepare_bar_counts(&lf, "carrier", BarOrder::Value, 2, &all_rows()).unwrap();
        assert_eq!(counts(&data), [(Some("AA"), 3.0), (Some("B6"), 2.0)]);
        assert_eq!(data.more, 2, "DL and UA are counted, not drawn");
        let data =
            prepare_bar_counts(&lf, "carrier", BarOrder::Value, BAR_CAP, &all_rows()).unwrap();
        assert_eq!(
            counts(&data)[2..],
            [(Some("DL"), 1.0), (Some("UA"), 1.0)],
            "ties A to Z"
        );

        let err = count_bars(&lf, "carrier", BarOrder::Value, BAR_CAP, 3, &all_rows())
            .unwrap_err()
            .to_string();
        assert_eq!(
            err,
            "more than 3 categories of carrier: counting stopped. Count by a column with \
             fewer values"
        );
        let data = count_bars(&lf, "carrier", BarOrder::Value, BAR_CAP, 4, &all_rows()).unwrap();
        assert_eq!(data.bars.len(), 4, "four is not more than four");
    }

    /// Batches are added up category by category, merged as they pile up; the read is
    /// told to stop as soon as the categories pass the cap.
    #[test]
    fn a_tally_merges_batches_and_stops_past_its_cap() {
        let batch = |ids: std::ops::Range<i64>| df!("id" => ids.collect::<Vec<_>>()).unwrap();
        let mut tally = Tally::new("id", 200_000);
        assert!(!tally.observe(&batch(0..70_000)).unwrap());
        assert_eq!(tally.merged, 70_000, "merged once the batches pile up");
        assert!(!tally.observe(&batch(0..10)).unwrap());
        let Counted::All { counts, rows } = tally.finish().unwrap() else {
            panic!("under the cap");
        };
        assert_eq!(rows, 70_010);
        let counts = counts.unwrap();
        assert_eq!(counts.height(), 70_000);
        let total: u64 = counts
            .column(COUNT_COLUMN)
            .unwrap()
            .u64()
            .unwrap()
            .sum()
            .unwrap();
        assert_eq!(total, 70_010);

        let mut tally = Tally::new("id", 1_000);
        assert!(
            tally.observe(&batch(0..70_000)).unwrap(),
            "past the cap: stop reading"
        );
        assert!(matches!(tally.finish().unwrap(), Counted::TooMany));
    }

    /// Through the streamed pass: past the cap the read stops and the count says so; a
    /// cancelled count is an error and is not held as the view's counts.
    #[test]
    fn a_streamed_count_stops_past_its_cap_or_when_cancelled() {
        let ids = df!("id" => (0..200_000i64).collect::<Vec<_>>())
            .unwrap()
            .lazy();
        let cancel = Arc::default();
        assert!(matches!(
            stream_counts(&ids, "id", 1_000, &cancel).unwrap(),
            Counted::TooMany
        ));

        let lf = species(30_000, 15_000, 4_999, 1);
        let sampling = ChartSampling::rows(Some(1_000));
        sampling.cancel.store(true, Ordering::Relaxed);
        let err = prepare_bar_counts(&lf, "species", BarOrder::Value, BAR_CAP, &sampling)
            .unwrap_err()
            .to_string();
        assert_eq!(err, "count cancelled");
        assert!(sampling.held.0.lock().unwrap().counts.is_empty());
        sampling.cancel.store(false, Ordering::Relaxed);
        let data = prepare_bar_counts(&lf, "species", BarOrder::Value, BAR_CAP, &sampling).unwrap();
        assert_eq!(data.counted, Some(50_000));
    }

    /// When the rows held are the whole view, Count counts them rather than reading
    /// again; a count is held too, so another order does not count again.
    #[test]
    fn counts_come_from_the_rows_held_and_are_held() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("flights.csv");
        std::fs::write(&path, "carrier,delay\nUA,1\nUA,2\nAA,3\n").unwrap();
        let lf = LazyCsvReader::new(PlRefPath::try_from_path(&path).unwrap())
            .finish()
            .unwrap();
        let sampling = all_rows();
        prepare_histogram_data(&lf, "delay", 10, ValueRange::All, &sampling).unwrap();
        std::fs::write(&path, "carrier,delay\nZZ,1\n").unwrap();
        // The rows held have no carrier: read, one count per category.
        let data = prepare_bar_counts(&lf, "carrier", BarOrder::Value, BAR_CAP, &sampling).unwrap();
        assert_eq!(counts(&data), [(Some("ZZ"), 1.0)]);

        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("flights.csv");
        std::fs::write(&path, "carrier,delay\nUA,1\nUA,2\nAA,3\n").unwrap();
        let lf = LazyCsvReader::new(PlRefPath::try_from_path(&path).unwrap())
            .finish()
            .unwrap();
        let sampling = all_rows();
        // Refused, carriers repeat; the rows it read stay held.
        assert!(
            prepare_bar_data(&lf, "carrier", "delay", BarOrder::Value, BAR_CAP, &sampling).is_err()
        );
        std::fs::write(&path, "carrier,delay\nZZ,1\n").unwrap();
        let data = prepare_bar_counts(&lf, "carrier", BarOrder::Value, BAR_CAP, &sampling).unwrap();
        assert_eq!(
            counts(&data),
            [(Some("UA"), 2.0), (Some("AA"), 1.0)],
            "counted from the rows held"
        );
        // Without the rows, only the count held can say this.
        sampling.held.0.lock().unwrap().rows = None;
        let data = prepare_bar_counts(&lf, "carrier", BarOrder::Label, BAR_CAP, &sampling).unwrap();
        assert_eq!(
            counts(&data),
            [(Some("AA"), 1.0), (Some("UA"), 2.0)],
            "another order from the count held"
        );

        // A view the sample size takes whole is read as a chart reads it, and its rows
        // held for the next chart.
        let sampling = ChartSampling {
            known_total: Some(3),
            ..all_rows()
        };
        std::fs::write(&path, "carrier,delay\nUA,1\nUA,2\nAA,3\n").unwrap();
        let data = prepare_bar_counts(&lf, "carrier", BarOrder::Value, BAR_CAP, &sampling).unwrap();
        assert_eq!(counts(&data), [(Some("UA"), 2.0), (Some("AA"), 1.0)]);
        let holding = sampling.held.0.lock().unwrap();
        let held = holding.rows.as_ref().expect("the rows read are held");
        assert_eq!(held.df.column("carrier").unwrap().len(), 3);
    }

    // ----- Aggregates, buckets, cumulative, color -----

    /// Daily rows of two symbols over three months, as a lazy frame: `date`,
    /// `symbol`, `ret` (A gains 1 a day, B 2).
    fn returns() -> LazyFrame {
        let days: Vec<i32> = (0..90).collect();
        let n = days.len();
        let mut df = df!(
            "date" => days.iter().chain(&days).map(|d| 19723 + d).collect::<Vec<i32>>(),
            "symbol" => std::iter::repeat_n("A", n).chain(std::iter::repeat_n("B", n)).collect::<Vec<_>>(),
            "ret" => std::iter::repeat_n(1.0, n).chain(std::iter::repeat_n(2.0, n)).collect::<Vec<f64>>()
        )
        .unwrap();
        df.apply("date", |c| c.cast(&DataType::Date).unwrap())
            .unwrap();
        df.lazy()
    }

    fn aggregate(
        lf: &LazyFrame,
        unit: crate::chart_modal::TimeUnit,
        aggregate: crate::chart_modal::Aggregate,
        cumulative: crate::chart_modal::Cumulative,
        color: Option<ColorSplit<'_>>,
    ) -> GroupedSeries {
        let schema = lf.clone().collect_schema().unwrap();
        let ys = ["ret".to_string()];
        prepare_aggregate_xy(
            lf,
            schema.as_ref(),
            &AggregateSpec {
                x: "date",
                time_unit: unit,
                ys: &ys,
                aggregate,
                quantile: 90,
                cumulative,
                color,
            },
            &all_rows(),
        )
        .unwrap()
    }

    /// A month bucket makes one point per month, the aggregate over every row in
    /// it, per color group, at the month's first day.
    #[test]
    fn a_time_bucket_aggregates_every_row_per_month_and_color() {
        use crate::chart_modal::{Aggregate, Cumulative, TimeUnit};
        let lf = returns();
        let groups = [Some("A".to_string()), Some("B".to_string())];
        let split = ColorSplit {
            column: "symbol",
            groups: &groups,
            other: false,
        };
        let sum = aggregate(
            &lf,
            TimeUnit::Month,
            Aggregate::Sum,
            Cumulative::Off,
            Some(split),
        );
        assert_eq!(sum.names, ["A", "B"]);
        assert_eq!(sum.x_axis_kind, XAxisTemporalKind::Date);
        assert_eq!(sum.rows.total_rows, 180, "every row, no sample");
        // 2024-01-01 is day 19723: January, February (29 days in 2024), March.
        let xs: Vec<f64> = sum.series[0].iter().map(|p| p.0).collect();
        assert_eq!(xs, [19723.0, 19754.0, 19783.0]);
        let a: Vec<f64> = sum.series[0].iter().map(|p| p.1).collect();
        let b: Vec<f64> = sum.series[1].iter().map(|p| p.1).collect();
        assert_eq!(a, [31.0, 29.0, 30.0]);
        assert_eq!(b, [62.0, 58.0, 60.0]);

        let mean = aggregate(
            &lf,
            TimeUnit::Month,
            Aggregate::Mean,
            Cumulative::Off,
            Some(split),
        );
        assert!(mean.series[1].iter().all(|p| p.1 == 2.0));
        let count = aggregate(
            &lf,
            TimeUnit::Quarter,
            Aggregate::Count,
            Cumulative::Off,
            None,
        );
        assert_eq!(count.names, ["count"]);
        assert_eq!(
            count.series[0],
            [(19723.0, 180.0)],
            "one quarter, both symbols"
        );
        let weeks = aggregate(&lf, TimeUnit::Week, Aggregate::Max, Cumulative::Off, None);
        // 2024-01-01 is a Monday: 90 days are 13 weeks less a day, in 13 buckets.
        assert_eq!(weeks.series[0].len(), 13);
        assert!(weeks.series[0].iter().all(|p| p.1 == 2.0));
    }

    /// Cumulative runs along X after the aggregate: a running sum, or returns
    /// compounded.
    #[test]
    fn cumulative_sums_or_compounds_along_x() {
        use crate::chart_modal::{Aggregate, Cumulative, TimeUnit};
        let lf = returns();
        let groups = [Some("A".to_string())];
        let split = ColorSplit {
            column: "symbol",
            groups: &groups,
            other: false,
        };
        let running = aggregate(
            &lf,
            TimeUnit::Month,
            Aggregate::Sum,
            Cumulative::Sum,
            Some(split),
        );
        let ys: Vec<f64> = running.series[0].iter().map(|p| p.1).collect();
        assert_eq!(ys, [31.0, 60.0, 90.0]);
        assert_eq!(running.names, ["A"], "only the groups picked");

        let mut points = vec![(0.0, 0.1), (1.0, 0.1), (2.0, -0.5)];
        accumulate(&mut points, Cumulative::Compound);
        let ys: Vec<f64> = points.iter().map(|p| (p.1 * 1e6).round() / 1e6).collect();
        assert_eq!(ys, [0.1, 0.21, -0.395]);
        let mut points = vec![(0.0, 3.0), (1.0, 4.0)];
        accumulate(&mut points, Cumulative::Off);
        assert_eq!(points, [(0.0, 3.0), (1.0, 4.0)]);
    }

    /// Compound runs over every row, not over a bucket's mean: 1% a day for 90 days,
    /// bucketed by month, is 1.01^31 - 1 at January's end, 1.01^60 - 1 at
    /// February's (29 days in 2024), 1.01^90 - 1 at March's.
    #[test]
    fn compound_runs_over_the_rows_of_each_bucket() {
        use crate::chart_modal::{Aggregate, Cumulative, TimeUnit};
        let mut df = df!(
            "date" => (0..90).map(|d| 19723 + d).collect::<Vec<i32>>(),
            "ret" => vec![0.01; 90]
        )
        .unwrap();
        df.apply("date", |c| c.cast(&DataType::Date).unwrap())
            .unwrap();
        let lf = df.lazy();
        for how in [Aggregate::Mean, Aggregate::Sum, Aggregate::Max] {
            let out = aggregate(&lf, TimeUnit::Month, how, Cumulative::Compound, None);
            let ys: Vec<f64> = out.series[0].iter().map(|p| p.1).collect();
            let want = [
                1.01f64.powi(31) - 1.0,
                1.01f64.powi(60) - 1.0,
                1.01f64.powi(90) - 1.0,
            ];
            for (y, w) in ys.iter().zip(want) {
                assert!((y - w).abs() < 1e-9, "{how:?}: {ys:?}");
            }
        }
        let rows = aggregate(
            &lf,
            TimeUnit::Month,
            Aggregate::Count,
            Cumulative::Compound,
            None,
        );
        let ys: Vec<f64> = rows.series[0].iter().map(|p| p.1).collect();
        assert_eq!(ys, [31.0, 60.0, 90.0], "a count runs as a count");
    }

    /// A group with no values is a gap, not the zero a sum of nothing is.
    #[test]
    fn a_bucket_with_no_values_is_a_gap() {
        use crate::chart_modal::{Aggregate, Cumulative, TimeUnit};
        let lf = df!(
            "date" => [0i32, 0, 1, 2],
            "ret" => [Some(1.0), Some(2.0), None, Some(4.0)]
        )
        .unwrap()
        .lazy()
        .with_column(col("date").cast(DataType::Date));
        let out = aggregate(&lf, TimeUnit::Day, Aggregate::Sum, Cumulative::Off, None);
        assert_eq!(out.series[0], [(0.0, 3.0), (2.0, 4.0)]);
        assert_eq!(out.breaks[0], [1], "the line breaks over day 1");
    }

    /// An X of nearly as many values as rows is refused before the group-by, which
    /// would hold a group per row.
    #[test]
    fn an_x_of_too_many_values_is_refused_first() {
        use crate::chart_modal::{Aggregate, Cumulative};
        let n = AGGREGATE_POINTS_MAX as i64 + 10_000;
        let lf = df!("x" => (0..n).collect::<Vec<i64>>(), "ret" => vec![1.0; n as usize])
            .unwrap()
            .lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let ys = ["ret".to_string()];
        let err = prepare_aggregate_xy(
            &lf,
            schema.as_ref(),
            &AggregateSpec {
                x: "x",
                time_unit: crate::chart_modal::TimeUnit::None,
                ys: &ys,
                aggregate: Aggregate::Mean,
                quantile: 90,
                cumulative: Cumulative::Off,
                color: None,
            },
            &all_rows(),
        )
        .unwrap_err();
        assert!(err.to_string().contains("values of x"), "{err}");
    }

    /// Color takes the values with the most rows, one per palette color, equal
    /// counts in the column's order; a pick takes its values, in the order picked.
    #[test]
    fn color_takes_the_largest_groups_or_the_ones_picked() {
        let values: Vec<String> = (0..9)
            .flat_map(|i| std::iter::repeat_n(format!("v{i}"), 10 + i))
            .chain(std::iter::once("v0".to_string()))
            .collect();
        let lf = df!("c" => values).unwrap().lazy();
        let rows = value_rows(&lf, "c", &all_rows()).unwrap();
        assert_eq!(rows.values.len(), 9);
        assert_eq!(rows.rows, 10 + 11 + 12 + 13 + 14 + 15 + 16 + 17 + 18 + 1);
        assert_eq!(rows.values[0], (Some("v8".to_string()), 18));
        let top = color_groups(&rows, &[], 7);
        assert_eq!(
            top,
            ["v8", "v7", "v6", "v5", "v4", "v3", "v2"]
                .map(|v| Some(v.to_string()))
                .to_vec()
        );
        // v0 has 11 rows, as many as v1: the column's order breaks the tie.
        assert_eq!(rows.values[7], (Some("v0".to_string()), 11));
        let picked = [Some("v1".to_string()), None];
        assert_eq!(color_groups(&rows, &picked, 7), picked);
        // A terminal of fewer colors draws fewer.
        assert_eq!(color_groups(&rows, &[], 3).len(), 3);
        assert_eq!(color_groups(&rows, &picked, 1), [Some("v1".to_string())]);
    }

    /// A line or scatter split by color without an aggregate: the sampled rows, a
    /// series per group in X order, rows of no group left out.
    #[test]
    fn a_color_splits_the_sampled_points() {
        let lf = df!(
            "x" => [3i64, 1, 2, 1, 2],
            "y" => [30.0, 10.0, 20.0, 1.0, 2.0],
            "c" => ["a", "a", "a", "b", "z"]
        )
        .unwrap()
        .lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let groups = [Some("a".to_string()), Some("b".to_string())];
        let split = ColorSplit {
            column: "c",
            groups: &groups,
            other: false,
        };
        let out = prepare_xy_by(&lf, schema.as_ref(), "x", "y", split, &all_rows()).unwrap();
        assert_eq!(out.names, ["a", "b"]);
        assert_eq!(out.series[0], [(1.0, 10.0), (2.0, 20.0), (3.0, 30.0)]);
        assert_eq!(out.series[1], [(1.0, 1.0)]);
    }

    /// Stdev, a quantile, first and last per X: the sample deviation (none for a
    /// group of one, so no point), a linearly interpolated percentile, and the first
    /// and last value in the rows' order, nulls passed over, which a sort sets.
    #[test]
    fn stdev_quantile_first_and_last_per_x() {
        use crate::chart_modal::{Aggregate, Cumulative, TimeUnit};
        // Read order is not value order: x=1 reads 4, 1, 3, 2.
        let lf = df!(
            "x" => [1i64, 1, 1, 1, 2, 3, 3],
            "y" => [Some(4.0), Some(1.0), Some(3.0), Some(2.0), Some(9.0), Some(5.0), None],
            "t" => [3i64, 1, 4, 2, 1, 2, 1],
            "c" => ["a", "b", "a", "b", "a", "a", "a"]
        )
        .unwrap()
        .lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let ys = ["y".to_string()];
        let run = |lf: &LazyFrame, aggregate, quantile| {
            let out = prepare_aggregate_xy(
                lf,
                schema.as_ref(),
                &AggregateSpec {
                    x: "x",
                    time_unit: TimeUnit::None,
                    ys: &ys,
                    aggregate,
                    quantile,
                    cumulative: Cumulative::Off,
                    color: None,
                },
                &all_rows(),
            )
            .unwrap();
            out.series[0].clone()
        };
        // x=1: 4, 1, 3, 2, mean 2.5, sample deviation sqrt(5/3).
        let stdev = run(&lf, Aggregate::Stdev, 90);
        assert_eq!(stdev.len(), 1, "x=2 and x=3 have one value each: {stdev:?}");
        assert!((stdev[0].1 - (5.0f64 / 3.0).sqrt()).abs() < 1e-12);
        // p90 of 1, 2, 3, 4: 3 + 0.7 = 3.7; p25: 1 + 0.75 = 1.75.
        let p90 = run(&lf, Aggregate::Quantile, 90);
        assert!((p90[0].1 - 3.7).abs() < 1e-12, "{p90:?}");
        assert_eq!(p90[1], (2.0, 9.0));
        let p25 = run(&lf, Aggregate::Quantile, 25);
        assert!((p25[0].1 - 1.75).abs() < 1e-12, "{p25:?}");
        // In read order: x=1 starts at 4 and ends at 2; x=3's last value is null,
        // so its last is the value before.
        assert_eq!(
            run(&lf, Aggregate::First, 90),
            [(1.0, 4.0), (2.0, 9.0), (3.0, 5.0)]
        );
        assert_eq!(
            run(&lf, Aggregate::Last, 90),
            [(1.0, 2.0), (2.0, 9.0), (3.0, 5.0)]
        );
        // Sorted by t: x=1 reads 1, 2, 4, 3.
        let sorted = lf.clone().sort(["t"], Default::default());
        assert_eq!(run(&sorted, Aggregate::First, 90)[0], (1.0, 1.0));
        assert_eq!(run(&sorted, Aggregate::Last, 90)[0], (1.0, 3.0));
        // A bar of the last per category, split by a color.
        let groups = [Some("a".to_string()), Some("b".to_string())];
        let bars = prepare_bar_aggregate(
            &lf,
            &BarAggregate {
                category: "x",
                value: Some("y"),
                aggregate: Aggregate::Last,
                quantile: 90,
                color: Some(ColorSplit {
                    column: "c",
                    groups: &groups,
                    other: false,
                }),
                order: BarOrder::Label,
                cap: BAR_CAP,
            },
            &all_rows(),
        )
        .unwrap();
        assert_eq!(bars.bars[0].by_group, [Some(3.0), Some(2.0)]);
        assert_eq!(bars.value_column, "last y");
        let p = prepare_bar_aggregate(
            &lf,
            &BarAggregate {
                category: "x",
                value: Some("y"),
                aggregate: Aggregate::Quantile,
                quantile: 90,
                color: None,
                order: BarOrder::Label,
                cap: BAR_CAP,
            },
            &all_rows(),
        )
        .unwrap();
        assert_eq!(p.value_column, "p90 y");
    }

    /// A distinct count of a string Y per X: nulls are no value, a group of only
    /// nulls is a gap; per color, and within a time bucket.
    #[test]
    fn distinct_counts_any_y_per_x() {
        use crate::chart_modal::{Aggregate, Cumulative, TimeUnit};
        let mut df = df!(
            "date" => [19723i32, 19723, 19723, 19724, 19724, 19754, 19755],
            "name" => [Some("Ann"), Some("Bo"), Some("Ann"), None, None, Some("Cy"), Some("Di")],
            "sex" => ["F", "M", "F", "F", "M", "M", "M"]
        )
        .unwrap();
        df.apply("date", |c| c.cast(&DataType::Date).unwrap())
            .unwrap();
        let lf = df.lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let ys = ["name".to_string()];
        let distinct = |unit, color| {
            prepare_aggregate_xy(
                &lf,
                schema.as_ref(),
                &AggregateSpec {
                    x: "date",
                    time_unit: unit,
                    ys: &ys,
                    aggregate: Aggregate::Distinct,
                    quantile: 90,
                    cumulative: Cumulative::Off,
                    color,
                },
                &all_rows(),
            )
            .unwrap()
        };
        let by_day = distinct(TimeUnit::Day, None);
        let ys_of = |s: &[(f64, f64)]| s.iter().map(|p| p.1).collect::<Vec<_>>();
        // Day 1: Ann, Bo; day 2: only nulls, a gap; then Cy, then Di.
        assert_eq!(ys_of(&by_day.series[0]), [2.0, 1.0, 1.0]);
        assert_eq!(by_day.breaks[0], [1], "the day of nulls breaks the line");
        let by_month = distinct(TimeUnit::Month, None);
        assert_eq!(ys_of(&by_month.series[0]), [2.0, 2.0], "Ann, Bo; Cy, Di");
        let groups = [Some("F".to_string()), Some("M".to_string())];
        let split = ColorSplit {
            column: "sex",
            groups: &groups,
            other: false,
        };
        let colored = distinct(TimeUnit::Month, Some(split));
        assert_eq!(colored.names, ["F", "M"]);
        assert_eq!(ys_of(&colored.series[0]), [1.0], "Ann");
        assert_eq!(ys_of(&colored.series[1]), [1.0, 2.0], "Bo; Cy, Di");
        // A bar of distinct names per sex: whole numbers.
        let bars = prepare_bar_aggregate(
            &lf,
            &BarAggregate {
                category: "sex",
                value: Some("name"),
                aggregate: Aggregate::Distinct,
                quantile: 90,
                color: None,
                order: BarOrder::Label,
                cap: BAR_CAP,
            },
            &all_rows(),
        )
        .unwrap();
        let values: Vec<f64> = bars.bars.iter().map(|b| b.value).collect();
        assert_eq!(values, [1.0, 3.0]);
        assert!(bars.value_dtype.is_integer());
        assert_eq!(bars.value_column, "distinct name");
    }

    /// With Other, every value without a group of its own (a null among them) is one
    /// more series, last: a scatter keeps every point, a line and a bar aggregate the
    /// rest as they do a group. Off, those rows are left out.
    #[test]
    fn other_gathers_every_value_without_a_series() {
        use crate::chart_modal::{Aggregate, Cumulative, TimeUnit};
        let lf = df!(
            "x" => [1i64, 1, 2, 2, 3, 3],
            "y" => [10.0, 1.0, 20.0, 2.0, 30.0, 4.0],
            "c" => [Some("a"), Some("b"), Some("a"), Some("z"), Some("a"), None]
        )
        .unwrap()
        .lazy();
        let schema = lf.clone().collect_schema().unwrap();
        let groups = [Some("a".to_string())];
        let split = |other| ColorSplit {
            column: "c",
            groups: &groups,
            other,
        };
        // Scatter.
        let out = prepare_xy_by(&lf, schema.as_ref(), "x", "y", split(true), &all_rows()).unwrap();
        assert_eq!(out.names, ["a", OTHER]);
        assert!(out.other);
        assert_eq!(out.series[1], [(1.0, 1.0), (2.0, 2.0), (3.0, 4.0)]);
        let out = prepare_xy_by(&lf, schema.as_ref(), "x", "y", split(false), &all_rows()).unwrap();
        assert_eq!(out.names, ["a"]);
        assert!(!out.other);
        // A line of the sum per x.
        let ys = ["y".to_string()];
        let line = |other| {
            prepare_aggregate_xy(
                &lf,
                schema.as_ref(),
                &AggregateSpec {
                    x: "x",
                    time_unit: TimeUnit::None,
                    ys: &ys,
                    aggregate: Aggregate::Sum,
                    quantile: 90,
                    cumulative: Cumulative::Off,
                    color: Some(split(other)),
                },
                &all_rows(),
            )
            .unwrap()
        };
        let on = line(true);
        assert_eq!(on.names, ["a", OTHER]);
        assert_eq!(on.series[1], [(1.0, 1.0), (2.0, 2.0), (3.0, 4.0)]);
        assert_eq!(on.rows.total_rows, 6, "every row is in a series");
        let off = line(false);
        assert_eq!(off.names, ["a"]);
        assert_eq!(off.rows.total_rows, 3);
        // A bar of the mean per x.
        let bars = |other| {
            let spec = BarAggregate {
                category: "x",
                value: Some("y"),
                aggregate: Aggregate::Mean,
                quantile: 90,
                color: Some(split(other)),
                order: BarOrder::Label,
                cap: BAR_CAP,
            };
            prepare_bar_aggregate(&lf, &spec, &all_rows()).unwrap()
        };
        let on = bars(true);
        assert_eq!(on.groups, ["a", OTHER]);
        assert!(on.other);
        let by: Vec<Vec<Option<f64>>> = on.bars.iter().map(|b| b.by_group.clone()).collect();
        assert_eq!(
            by,
            [
                vec![Some(10.0), Some(1.0)],
                vec![Some(20.0), Some(2.0)],
                vec![Some(30.0), Some(4.0)]
            ]
        );
        let off = bars(false);
        assert_eq!(off.groups, ["a"]);
        assert!(off.bars.iter().all(|b| b.by_group.len() == 1));
    }

    /// A bar of the mean per category, split by color: a value per group in each
    /// bar, ordered by the largest group; a count needs no value column.
    #[test]
    fn bars_aggregate_per_category_and_color() {
        use crate::chart_modal::Aggregate;
        let lf = df!(
            "carrier" => ["UA", "UA", "UA", "AA", "AA"],
            "origin" => ["EWR", "EWR", "JFK", "EWR", "JFK"],
            "delay" => [10.0, 20.0, 5.0, 1.0, 50.0]
        )
        .unwrap()
        .lazy();
        let groups = [Some("EWR".to_string()), Some("JFK".to_string())];
        let split = ColorSplit {
            column: "origin",
            groups: &groups,
            other: false,
        };
        let spec = BarAggregate {
            category: "carrier",
            value: Some("delay"),
            aggregate: Aggregate::Mean,
            quantile: 90,
            color: Some(split),
            order: BarOrder::Value,
            cap: BAR_CAP,
        };
        let data = prepare_bar_aggregate(&lf, &spec, &all_rows()).unwrap();
        assert_eq!(data.groups, ["EWR", "JFK"]);
        assert_eq!(data.value_column, "mean delay");
        assert_eq!(data.rows.total_rows, 5);
        let bars: Vec<(Option<&str>, Vec<Option<f64>>)> = data
            .bars
            .iter()
            .map(|b| (b.label.as_deref(), b.by_group.clone()))
            .collect();
        assert_eq!(
            bars,
            [
                (Some("AA"), vec![Some(1.0), Some(50.0)]),
                (Some("UA"), vec![Some(15.0), Some(5.0)])
            ],
            "AA's largest group is larger"
        );
        let count = BarAggregate {
            value: None,
            aggregate: Aggregate::Count,
            quantile: 90,
            ..spec
        };
        let data = prepare_bar_aggregate(&lf, &count, &all_rows()).unwrap();
        assert_eq!(data.bars[0].label.as_deref(), Some("UA"));
        assert_eq!(data.bars[0].value, 3.0, "a count adds up across groups");
        assert!(data.value_dtype.is_integer(), "counts print whole");
        // A NaN draws nothing; a category whose values are all null has no bar.
        let odd = df!(
            "carrier" => ["UA", "AA", "DL"],
            "delay" => [Some(f64::NAN), None, Some(1.0)]
        )
        .unwrap()
        .lazy();
        let mean = BarAggregate {
            color: None,
            ..spec
        };
        let data = prepare_bar_aggregate(&odd, &mean, &all_rows()).unwrap();
        let labels: Vec<Option<&str>> = data.bars.iter().map(|b| b.label.as_deref()).collect();
        assert_eq!(labels, [Some("DL")]);
        assert_eq!(data.no_value, 2);
        let sum = BarAggregate {
            aggregate: Aggregate::Sum,
            quantile: 90,
            color: None,
            ..spec
        };
        let data = prepare_bar_aggregate(&lf, &sum, &all_rows()).unwrap();
        assert_eq!(
            data.bars
                .iter()
                .map(|b| (b.label.as_deref(), b.value))
                .collect::<Vec<_>>(),
            [(Some("AA"), 51.0), (Some("UA"), 35.0)]
        );
    }

    /// A histogram split by color: every group on the same bins, each a share of
    /// its own rows when asked.
    #[test]
    fn a_histogram_splits_into_groups_on_shared_bins() {
        let lf = df!(
            "v" => [0.0, 1.0, 2.0, 3.0, 0.0, 0.0],
            "g" => ["a", "a", "a", "a", "b", "b"]
        )
        .unwrap()
        .lazy();
        let groups = [Some("a".to_string()), Some("b".to_string())];
        let split = ColorSplit {
            column: "g",
            groups: &groups,
            other: false,
        };
        let data =
            prepare_histogram_by(&lf, "v", 3, ValueRange::All, true, Some(split), &all_rows())
                .unwrap();
        assert_eq!(data.bins.len(), 3);
        assert_eq!(data.groups.len(), 2);
        assert_eq!(data.groups[0].counts, [0.25, 0.25, 0.5]);
        assert_eq!(data.groups[1].counts, [1.0, 0.0, 0.0]);
        assert_eq!(data.max_count, 1.0);
        let total: f64 = data.bins.iter().map(|b| b.count).sum();
        assert!((total - 1.0).abs() < 1e-9, "the whole is a share too");
    }

    /// A box per category, the categories given.
    #[test]
    fn a_box_per_category() {
        let lf = df!(
            "v" => [1.0, 2.0, 3.0, 10.0, 20.0],
            "k" => ["x", "x", "x", "y", "y"]
        )
        .unwrap()
        .lazy();
        let groups = [Some("y".to_string()), Some("x".to_string())];
        let split = ColorSplit {
            column: "k",
            groups: &groups,
            other: false,
        };
        let data = prepare_box_by(&lf, "v", split, ValueRange::All, &all_rows()).unwrap();
        let names: Vec<&str> = data.stats.iter().map(|s| s.name.as_str()).collect();
        assert_eq!(names, ["y", "x"]);
        assert_eq!(data.stats[1].median, 2.0);
        assert_eq!((data.y_min, data.y_max), (1.0, 20.0));
    }
}