// The index-generic reduction kernel, and the **canonical** statement of the
// null / `NaN` rules every column reduction follows.
//
// `Series::sum` / `mean` / `min` / `max`, the grouped `agg`
// reducers, and the scoped `eval_agg` all route through the factories here
// rather than carrying null-skipping / `NaN`-handling folds of their own: each
// resolves the dtype dispatch and (for the `Builtin` backend) the validity
// mask **once**, returning a total fold that reduces an index window — a count
// `n` plus an indexing function `at` mapping `0 ..< n` onto source rows — to a
// single `Scalar`. The whole-column callers fold `(len, k => k)`; the per-group
// caller hoists the factory once and folds each group `(g.length(), k => g[k])`,
// so the grouped reduction stays a single linear pass over the column rather
// than re-materialising the validity mask per group.
//
// The expression layer's aggregation family (`Expr::std` / `variance` / `median` /
// `n_unique` / `first` / `last`) reduces through the same kernel: each is a
// window reducer here, so it serves the whole-column, per-group, and per-scope
// surfaces with one fold and one set of null / `NaN` rules.
//
// The rules, canonically: `NaN` is a value, only `Null` is missing.
// `sum` / `mean` add `NaN` in so it propagates — and `var` / `std`, built on
// the mean, propagate it too; `median`, an order statistic, propagates it the
// same way. The extremum fold skips it (matching Polars' regular
// `min` / `max`, the one value-following fold that does), and `sort` orders
// it by the IEEE total order — larger than every other value, so it trails an
// ascending sort and leads a descending one. `n_unique` keys a `NaN` into one
// bucket (it is one value), and `first` / `last` are positional, so they skip
// nothing — a present `NaN` is returned verbatim.
//
// One qualification, since "canonical" is not "only": four of the `Series`
// reductions — `sum` / `mean` / `min` / `max` — dispatch a `Numeric`-backed
// column to the bitmap-free fold beside the fast-path layout
// before reaching here, and that fold implements the empty-column and `NaN`
// rules above a second time. So these are the rules to *read*, and to change
// first, but not the only code that encodes them: the two are held cell-for-cell
// in step by the differential parity harness, which runs every
// backend-sensitive `Series` op on both representations of the same column and
// asserts the results identical. The remaining reductions take no such fast
// path — this kernel already skips the validity materialisation for an all-valid
// `Numeric` column (`reduce_present` hands back a constant-`true` predicate),
// which is most of what the fast path buys.

///|
/// Per-row presence predicate, hoisted once per reduction. A `Builtin` column
/// materialises its validity mask; an all-valid `Numeric` column needs none
/// (every slot is present), so the predicate is constant `true` and no bitmap
/// is allocated. Read at a *source* row index, which the reducers reach through
/// their `at` mapping.
fn reduce_present(storage : @column.ColumnStorage) -> (Int) -> Bool {
  match storage.kind() {
    @column.StorageKind::Numeric => _ => true
    @column.StorageKind::Builtin => {
      let valid = storage.validity().to_bools()
      i => valid[i]
    }
  }
}

///|
/// `sum` / `mean` reducer over an index window. The accumulation is the
/// shared folds in `internal/numeric` — `fold_int64_sum` (the exact `Int64`
/// sum, wrapping past 2^63) and `fold_double_sum` (the `Double` accumulator
/// that widens an `Int` cell before adding, so a mean cannot wrap, and that
/// sums a present `NaN` in, so it propagates) — whose docstrings state those
/// rules once; this reducer adds the window mapping and the empty-window
/// translation. With `want_mean` the running total divides by the non-null
/// count and is `Scalar::Null` for an empty / all-null window (the surfaces
/// translate that: `Series::mean` to `InvalidOperation`, the grouped / scoped
/// reducers to a null cell); otherwise the result is the dtype-preserving sum,
/// the additive identity for an empty / all-null window. A non-numeric
/// column (the registry numericity flag: `Bool` / `String`, and an I64-riding
/// dtype like `Date`) `raise`s `TypeMismatch` here — at factory time,
/// before any window is folded — matching the previous per-surface reducers,
/// which rejected the dtype up front (and so reject it even for zero groups).
fn reduce_arith(
  storage : @column.ColumnStorage,
  want_mean~ : Bool,
) -> ((Int, (Int) -> Int) -> @types.Scalar) raise @types.DataError {
  let present = reduce_present(storage)
  match storage.data() {
    // The dtype gate, not the buffer variant, decides numericity: a Date
    // column is I64-typed but is not a number, and summing day-counts is a
    // bug, not a reduction. Non-numeric columns raise at factory time, before
    // any window is folded.
    @column.ColumnData::I64(a) if storage.dtype().is_numeric() =>
      if want_mean {
        // The mean's output is `Double`, so the accumulator widens each cell
        // before adding (`fold_double_sum`) — the mean-safe form the shared
        // fold documents; a count of 0 is the empty / all-null window.
        fn(n : Int, at : (Int) -> Int) {
          let (acc, cnt) = @numeric.fold_double_sum(
            n,
            k => a[at(k)].to_double(),
            k => present(at(k)),
          )
          if cnt == 0 {
            @types.Scalar::Null
          } else {
            @types.Scalar::Float(acc / cnt.to_double())
          }
        }
      } else {
        // `sum` keeps the `Int` dtype, so it stays the exact `Int64` fold
        // (`fold_int64_sum`) — sums past 2^63 overflow, the documented price
        // of an integer sum.
        fn(n : Int, at : (Int) -> Int) {
          @types.Scalar::Int(
            @numeric.fold_int64_sum(n, k => a[at(k)], k => present(at(k))),
          )
        }
      }
    @column.ColumnData::F64(a) =>
      fn(n : Int, at : (Int) -> Int) {
        // `NaN` is a value, not missing: it participates and propagates to a
        // `NaN` total; only a null cell (`!present`) is skipped — the shared
        // `fold_double_sum` rule.
        let (acc, cnt) = @numeric.fold_double_sum(n, k => a[at(k)], k => {
          present(at(k))
        })
        if want_mean {
          if cnt == 0 {
            @types.Scalar::Null
          } else {
            @types.Scalar::Float(acc / cnt.to_double())
          }
        } else {
          @types.Scalar::Float(acc)
        }
      }
    @column.ColumnData::I64(_)
    | @column.ColumnData::Bool(_)
    | @column.ColumnData::Utf8(_) => {
      let verb = if want_mean { "mean" } else { "sum" }
      raise @types.DataError::TypeMismatch(
        @types.TypeMismatchDetail::Message(
          "\{verb} undefined for \{storage.dtype()}",
        ),
      )
    }
  }
}

///|
/// `min` / `max` reducer over an index window (`want_min` picks which),
/// keeping the source dtype. Total over every dtype — `Bool` orders
/// `false < true`, `String` by `compare_string_lex` — returning the dtype's
/// `Scalar` variant or `Scalar::Null` for an empty / all-null / all-`NaN`
/// (`Float`) window. `NaN` is treated as missing (skipped), as in Polars'
/// regular `min` / `max` (`sort` instead orders `NaN` as the largest value);
/// only a null slot is otherwise skipped. Shares the single `fold_extremum`
/// accumulation.
fn reduce_extremum(
  storage : @column.ColumnStorage,
  want_min~ : Bool,
) -> (Int, (Int) -> Int) -> @types.Scalar {
  let present = reduce_present(storage)
  match storage.data() {
    @column.ColumnData::I64(a) =>
      fn(n : Int, at : (Int) -> Int) {
        let get = k => a[at(k)]
        let skip = k => !present(at(k))
        let better = (x, y) => if want_min { x < y } else { x > y }
        match @numeric.fold_extremum(n, get, skip, better) {
          // The extremum lifts through the dtype via the registry's rider
          // lift (`i64_rider_scalar`): a Date column min/maxes to a Date,
          // not to the day count its I64 buffer holds.
          Some(b) =>
            match storage.dtype().i64_rider_scalar(b) {
              Some(s) => s
              None => @types.Scalar::Int(b)
            }
          None => @types.Scalar::Null
        }
      }
    @column.ColumnData::F64(a) =>
      fn(n : Int, at : (Int) -> Int) {
        let get = k => a[at(k)]
        // `NaN` is treated as missing (skipped), as in `Series::min`.
        let skip = k => !present(at(k)) || a[at(k)].is_nan()
        let better = (x, y) => if want_min { x < y } else { x > y }
        match @numeric.fold_extremum(n, get, skip, better) {
          Some(b) => @types.Scalar::Float(b)
          None => @types.Scalar::Null
        }
      }
    @column.ColumnData::Bool(a) =>
      fn(n : Int, at : (Int) -> Int) {
        let get = k => a[at(k)]
        let skip = k => !present(at(k))
        // `false < true`: under min `false` beats `true`, under max reverse.
        let better = (x, y) => if want_min { !x && y } else { x && !y }
        match @numeric.fold_extremum(n, get, skip, better) {
          Some(b) => @types.Scalar::Bool(b)
          None => @types.Scalar::Null
        }
      }
    @column.ColumnData::Utf8(a) =>
      fn(n : Int, at : (Int) -> Int) {
        let get = k => a.get(at(k))
        let skip = k => !present(at(k))
        let better = (x, y) => {
          if want_min {
            @text.compare_string_lex(x, y) < 0
          } else {
            @text.compare_string_lex(x, y) > 0
          }
        }
        match @numeric.fold_extremum(n, get, skip, better) {
          Some(b) => @types.Scalar::String(b)
          None => @types.Scalar::Null
        }
      }
  }
}

///|
/// `count` reducer over an index window: the number of non-null cells, as a
/// `Scalar::Int`. Dtype-independent (it reads only the presence predicate), so
/// it needs no per-dtype dispatch and is total over every column.
fn reduce_count(
  storage : @column.ColumnStorage,
) -> (Int, (Int) -> Int) -> @types.Scalar {
  let present = reduce_present(storage)
  fn(n : Int, at : (Int) -> Int) {
    let mut c : Int64 = 0
    for k in 0.. Double` cell
/// reader: an `Int` cell widens through `to_double`, a `Float` cell reads
/// directly. Shared by the numeric-only reductions `variance` / `std` and
/// `median`, which otherwise repeat this same three-arm match. A non-numeric
/// column (registry numericity: `Bool` / `String`, and an I64-riding dtype
/// like `Date`) has no such reading, so it `raise`s `TypeMismatch`
/// here — naming the operation (`op`) that asked — at factory time, before any
/// window is folded, exactly as `reduce_arith` rejects a non-numeric column.
fn numeric_double_reader(
  storage : @column.ColumnStorage,
  op : String,
) -> ((Int) -> Double) raise @types.DataError {
  match storage.data() {
    @column.ColumnData::I64(a) if storage.dtype().is_numeric() =>
      i => a[i].to_double()
    @column.ColumnData::F64(a) => i => a[i]
    @column.ColumnData::I64(_)
    | @column.ColumnData::Bool(_)
    | @column.ColumnData::Utf8(_) =>
      raise @types.DataError::TypeMismatch(
        @types.TypeMismatchDetail::Message(
          "\{op} undefined for \{storage.dtype()}",
        ),
      )
  }
}

///|
/// `var` / `std` reducer over an index window (`want_std` square-roots the
/// result): the **sample** statistics with `ddof = 1` (Polars' default). The
/// variance is `Σ(xᵢ − mean)² / (cnt − 1)` over the `cnt` non-null cells and
/// the standard deviation its square root, both `Scalar::Float` — `Int` widens
/// to `Double`, so the output always types `Float`. Computed by Welford's
/// online algorithm (a single pass that never forms the raw sum), so a
/// finite-variance window of near-`Double`-max values — whose sum would
/// overflow to `±inf` and poison a two-pass mean, reporting `+inf` for a true
/// variance of 0 — still yields its finite variance. When the true variance
/// itself exceeds `Double`'s range (finite cells more than `Double::MAX`
/// apart, e.g. opposite signs near `±1.8e308`), the result saturates to
/// `+inf` — never a negative variance or a `NaN` std from the internal
/// overflow — matching numpy / Polars. Numeric only: a
/// non-numeric column (registry numericity: `Bool` / `String`, and an
/// I64-riding dtype like `Date`) `raise`s `TypeMismatch` here at
/// factory time, before any window is folded, exactly as `reduce_arith` does.
/// A `NaN` is a present value: when at least two non-null cells are present it
/// counts toward `cnt` and **propagates** — the mean is `NaN`, hence every
/// deviation and the result — matching `sum` / `mean`. The fewer-than-two rule
/// takes precedence, though: a window with fewer than two non-null cells has no
/// sample variance (the `cnt − 1` denominator is zero or negative), so it
/// reduces to `Scalar::Null` (the empty / single-value / all-null cases — so a
/// lone `NaN` cell is `Null`, not `NaN`), which the surfaces translate the way
/// they do `mean`'s null cell.
fn reduce_variance(
  storage : @column.ColumnStorage,
  want_std~ : Bool,
) -> ((Int, (Int) -> Int) -> @types.Scalar) raise @types.DataError {
  let present = reduce_present(storage)
  let read = numeric_double_reader(
    storage,
    if want_std {
      "std"
    } else {
      "variance"
    },
  )
  fn(n : Int, at : (Int) -> Int) {
    // Welford's online algorithm: one pass that never forms the raw sum, so a
    // finite-variance window of near-`Double`-max values (whose sum would
    // overflow to `±inf` and poison a two-pass mean, turning a true variance of
    // 0 into `+inf`) still yields its finite variance. `mean` is the running
    // mean, `m2` the running sum of squared deviations from it.
    let mut cnt = 0
    let mut mean = 0.0
    let mut m2 = 0.0
    let mut saw_nonfinite = false
    for k in 0..= 0` under finite in-range arithmetic, but `delta`
      // itself overflows to `±inf` when two finite cells sit more than
      // `Double::MAX` apart (opposite signs near `±1.8e308`), and the
      // artefact poisons `m2` to `-inf` or `NaN` — reporting a mathematically
      // impossible negative variance and a `NaN` std. With every *input*
      // finite, a negative/`NaN` `m2` can only be that artefact, so saturate
      // to `+inf`: the true variance exceeds `Double`'s range (numpy/Polars
      // report `+inf` here too), and `sqrt(+inf) = +inf` keeps std
      // consistent. Genuine `NaN` / `±inf` inputs skip the rescue and keep
      // IEEE propagation, matching `sum` / `mean`.
      let m2 = if !saw_nonfinite && !(m2 >= 0.0) {
        @double.infinity
      } else {
        m2
      }
      let variance = m2 / (cnt - 1).to_double()
      @types.Scalar::Float(if want_std { variance.sqrt() } else { variance })
    }
  }
}

///|
/// `median` reducer over an index window: the middle of the sorted non-null
/// cells, or the mean of the two middles for an even count, as a
/// `Scalar::Float` — `Int` widens to `Double`, so a median always types
/// `Float` (Polars). Numeric only: a non-numeric column `raise`s
/// `TypeMismatch` at factory time, as `reduce_arith` does. A `NaN` is
/// **propagated** — the `sum` / `mean` rule, not the `min` / `max` skip: a
/// window that contains one has that value's order statistic. An empty /
/// all-null window reduces to `Scalar::Null`.
fn reduce_median(
  storage : @column.ColumnStorage,
) -> ((Int, (Int) -> Int) -> @types.Scalar) raise @types.DataError {
  let present = reduce_present(storage)
  let read = numeric_double_reader(storage, "median")
  fn(n : Int, at : (Int) -> Int) {
    // Collect the present values, then sort: the median is the middle by
    // order. A `NaN` anywhere in the window is the whole answer, so it is
    // detected before the middles are read.
    let vals : Array[Double] = []
    let mut saw_nan = false
    for k in 0.. (Int, (Int) -> Int) -> Int {
  let present = reduce_present(storage)
  let key_of : (Int) -> KeyCell = match storage.data() {
    @column.ColumnData::I64(a) => i => KInt(a[i])
    // The `Float` arm shares `float_key_cell` with `key_cell` (NaN → KNaN,
    // -0.0 → +0.0) so it keys on the native `Double` without boxing a `Scalar`
    // per cell, like the other three arms — n_unique / group_by / join stay in
    // agreement on "distinct" by construction, not by hand-synced copies.
    @column.ColumnData::F64(a) => i => float_key_cell(a[i])
    @column.ColumnData::Bool(a) => i => KBool(a[i])
    @column.ColumnData::Utf8(a) => i => KStr(a.get(i))
  }
  fn(n : Int, at : (Int) -> Int) {
    let seen : Map[KeyCell, Unit] = Map([])
    for k in 0.. (Int, (Int) -> Int) -> @types.Scalar {
  let present = reduce_present(storage)
  // Resolve the dtype dispatch once: a reader that builds the source-dtype
  // `Scalar` at a source row. Only ever called on a present row (the null /
  // empty cases short-circuit to `Null` first), so it never reads a null slot.
  // The read lifts through the dtype registry, so an I64-riding dtype
  // (Date) surfaces its own Scalar rather than the buffer's Int.
  let dt = storage.dtype()
  let data = storage.data()
  let read : (Int) -> @types.Scalar = i => @column.scalar_of(dt, data, i)
  fn(n : Int, at : (Int) -> Int) {
    if n == 0 {
      @types.Scalar::Null
    } else {
      let i = at(if want_first { 0 } else { n - 1 })
      if present(i) {
        read(i)
      } else {
        @types.Scalar::Null
      }
    }
  }
}

///|
/// The eleven reductions the shared kernel resolves — the single aggregation
/// vocabulary, one fact with one home. The expression AST's `Agg` node carries
/// a `ReduceOp` directly (the AST package imports `series`), `frame`'s evaluator
/// and grouped-`agg` fast path dispatch on it, and `reducer_for` right below
/// resolves it — so adding a reduction is one variant here plus its
/// `reducer_for` arm and the public `Expr` builder, with no mirror enum and no
/// translation layer to keep in lockstep. `pub(all)` so the engine packages
/// can construct it.
#doc(hidden)
#internal(engine, "MoonFrame execution engine API")
pub(all) enum ReduceOp {
  Sum
  Mean
  Min
  Max
  Count
  Std
  Var
  Median
  NUnique
  First
  Last
} derive(Eq)

///|
/// Equality for the engine packages: the expression AST carries a `ReduceOp`
/// inside its `Agg` node, and the AST's own `Eq` derive reaches this method
/// from another package. The attributes keep the promoted methods out of the
/// public interface, like the enum they belong to.
#doc(hidden)
#internal(engine, "MoonFrame execution engine API")
pub extend ReduceOp with Eq::{equal, not_equal}

///|
/// Resolve a `ReduceOp` to its window reducer plus the dtype probe that fixes
/// the output column's type: `Count` / `NUnique` are always `Int`, `Mean` /
/// `Std` / `Var` / `Median` always `Float`, and `Sum` / `Min` / `Max` /
/// `First` / `Last` keep the source dtype. Shared by the grouped (`agg`) and
/// scoped (`eval_agg`) reductions so the op → reducer mapping lives in one
/// place. `Sum` / `Mean` / `Std` / `Var` / `Median` `raise` `TypeMismatch`
/// here for a non-numeric column, before any window work.
///
/// The probe is a logical `DataType`, not a physical witness column: the
/// caller (`frame`'s aggregation layer) only forwards it back into
/// `scalars_to_series`, and forwarding a dtype keeps the storage
/// representation on this side of the seam.
#doc(hidden)
#internal(engine, "MoonFrame execution engine API")
pub fn Series::reducer_for(
  self : Series,
  op : ReduceOp,
) -> ((Int, (Int) -> Int) -> @types.Scalar, @types.DataType) raise @types.DataError {
  let storage = self.storage
  match op {
    Count => (reduce_count(storage), @types.DataType::Int)
    Sum => (reduce_arith(storage, want_mean=false), storage.dtype())
    Mean => (reduce_arith(storage, want_mean=true), @types.DataType::Float)
    Min => (reduce_extremum(storage, want_min=true), storage.dtype())
    Max => (reduce_extremum(storage, want_min=false), storage.dtype())
    Std => (reduce_variance(storage, want_std=true), @types.DataType::Float)
    Var => (reduce_variance(storage, want_std=false), @types.DataType::Float)
    Median => (reduce_median(storage), @types.DataType::Float)
    NUnique => {
      // The `n_unique` reducer over an index window: distinct non-null count
      // as `Scalar::Int`, total over every dtype and never null (an empty /
      // all-null window counts zero), over the shared `count_distinct` core
      // `Series::n_unique` also uses.
      let counter = count_distinct(storage)
      (
        fn(n : Int, at : (Int) -> Int) {
          @types.Scalar::Int(counter(n, at).to_int64())
        },
        @types.DataType::Int,
      )
    }
    First => (reduce_first_last(storage, want_first=true), storage.dtype())
    Last => (reduce_first_last(storage, want_first=false), storage.dtype())
  }
}

///|
/// Project per-window reduction `Scalar`s into a typed output column: the
/// per-dtype projection is the column layer's `BuiltinColumn::from_scalars`
/// (the bulk write side of the identity/representation meet, whose read side
/// is `scalar_of`), and this wrapper adds what a `Series` needs — the column
/// name and the all-valid numeric convergence onto the `Numeric` fast path
/// (`try_column_to_numeric`, a no-op for a nullable, `Bool`, or `String`
/// column, which keeps `Builtin`). This is the single column builder behind
/// the grouped (`agg`) and scoped (`eval_agg`) reductions, so both agree on
/// backend as well as on every cell value: the projection's catch-all `None`
/// arm is reached by the nullable reductions (`mean` / `min` / `max` / `std` /
/// `var` / `median` over an empty / all-null window, and `first` / `last`
/// over a null cell), so no arm is dead.
///
/// `probe` is the logical dtype the caller resolved (`reducer_for`'s second
/// result, the cells' own inferred dtype, or an empty-scope evaluation's), so
/// no caller outside this package has to name a storage type to build a
/// column. `Null` is not a column dtype — it is a value-level concept with no
/// physical backend — so it `raise`s `Unsupported`, exactly as
/// `Series::empty_of` does.
#doc(hidden)
#internal(engine, "MoonFrame execution engine API")
pub fn scalars_to_series(
  probe : @types.DataType,
  name : String,
  cells : Array[@types.Scalar],
) -> Series raise @types.DataError {
  Series::from_builtin(
    name,
    @column.BuiltinColumn::from_scalars(probe, cells, name~),
  ).try_column_to_numeric()
}