// 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 (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) -> (Int32) -> 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. The decimal arm of `sum` carries the checked twin and
/// refuses on overflow, so the reducer's type raises. 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,
) -> ((Int32, (Int32) -> Int32) -> @types.Scalar raise @types.DataError) 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.
    // The summable gate: `sum` also takes the exact I64 riders (a Decimal
    // column of one scale adds exactly), while `mean` — an approximate,
    // `Double`-valued statistic — keeps the numeric gate, so a summable but
    // non-numeric dtype raises here at factory time exactly as a
    // non-summable one does below.
    @column.ColumnData::I64(a) if storage.dtype().is_summable() => {
      if want_mean && !storage.dtype().is_numeric() {
        raise @types.DataError::TypeMismatch(
          @types.TypeMismatchDetail::Message(
            "mean undefined for \{storage.dtype()}: the mean is approximate (a Double), so it stays fenced to numeric dtypes — sum is the exact form",
          ),
        )
      }
      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 : Int32, at : (Int32) -> Int32) {
          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())
          }
        }
        // `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.
      } else if storage.dtype() is @types.DataType::Decimal(scale) {
        // The decimal sum is the checked twin of the exact `Int64` fold
        // over the unscaled cells, re-wrapped at the column's scale: a
        // total past the `Int64` range refuses with `InvalidOperation` —
        // the exact dtype's contract, where the `Int` arm's wraparound is
        // a price only the wrapping dtype agreed to.
        fn(
          n : Int32,
          at : (Int32) -> Int32,
        ) -> @types.Scalar raise @types.DataError {
          match
            @numeric.fold_int64_sum_checked(n, k => a[at(k)], k => {
              present(at(k))
            }) {
            Some(total) => @types.Scalar::Decimal(scale, total)
            None =>
              raise @types.DataError::InvalidOperation(
                "decimal sum overflows the 18-digit magnitude",
              )
          }
        }
      } else {
        fn(n : Int32, at : (Int32) -> Int32) {
          @types.Scalar::Int(
            @numeric.fold_int64_sum(n, k => a[at(k)], k => present(at(k))),
          )
        }
      }
    }
    @column.ColumnData::F64(a) =>
      fn(n : Int32, at : (Int32) -> Int32) {
        // `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; `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,
) -> (Int32, (Int32) -> Int32) -> @types.Scalar {
  let present = reduce_present(storage)
  match storage.data() {
    @column.ColumnData::I64(a) =>
      fn(n : Int32, at : (Int32) -> Int32) {
        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 : Int32, at : (Int32) -> Int32) {
        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 : Int32, at : (Int32) -> Int32) {
        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 : Int32, at : (Int32) -> Int32) {
        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,
) -> (Int32, (Int32) -> Int32) -> @types.Scalar {
  let present = reduce_present(storage)
  fn(n : Int32, at : (Int32) -> Int32) {
    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,
) -> ((Int32) -> 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`. 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 — `@numeric.fold_welford_m2`, the one fold this reducer
/// and the rolling kernel share — 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, the fold saturates to `+inf` — never a
/// negative variance or a `NaN` std from the internal overflow. 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,
) -> ((Int32, (Int32) -> Int32) -> @types.Scalar raise @types.DataError) raise @types.DataError {
  let present = reduce_present(storage)
  let read = numeric_double_reader(
    storage,
    if want_std {
      "std"
    } else {
      "variance"
    },
  )
  fn(n : Int32, at : (Int32) -> Int32) {
    // Welford's fold over the window's present cells — the shared
    // `@numeric` primitive, saturation rescue included (its docstring owns
    // the algorithm's overflow semantics).
    let (m2, cnt, _) = @numeric.fold_welford_m2(n, k => read(at(k)), k => {
      present(at(k))
    })
    // The sample denominator `cnt - 1` is zero or negative for an empty,
    // single-value, or all-null window — no sample variance, a null cell (so a
    // lone `NaN` cell is `Null`, not `NaN`); the fewer-than-two rule precedes
    // NaN propagation.
    if cnt < 2 {
      @types.Scalar::Null
    } else {
      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`. 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,
) -> ((Int32, (Int32) -> Int32) -> @types.Scalar raise @types.DataError) raise @types.DataError {
  let present = reduce_present(storage)
  let read = numeric_double_reader(storage, "median")
  fn(n : Int32, at : (Int32) -> Int32) {
    // 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.. ((Int32, (Int32) -> Int32) -> @types.Scalar raise @types.DataError) raise @types.DataError {
  // `NaN` fails both comparisons, so one conjunction rejects it with the
  // out-of-range values rather than needing its own test.
  if !(q >= 0.0 && q <= 1.0) {
    raise @types.DataError::InvalidOperation("quantile q must lie in [0, 1]")
  }
  let present = reduce_present(storage)
  let read = numeric_double_reader(storage, "quantile")
  fn(n : Int32, at : (Int32) -> Int32) {
    // Collect the present values, then sort: the quantile reads the sorted
    // order at pos = q * (m - 1). A `NaN` anywhere in the window is the whole
    // answer, so it is detected before positions are read — the median rule.
    let vals : Array[Double] = []
    let mut saw_nan = false
    for k in 0..
                if pos - lo_idx.to_double() >= 0.5 {
                  higher
                } else {
                  lower
                }
              Linear => {
                let frac = pos - lo_idx.to_double()
                if frac == 0.0 {
                  lower
                } else {
                  // The convex combination, not `lower + frac * (higher -
                  // lower)`: the difference of opposite-sign neighbours
                  // adds their magnitudes and can round to `+inf` before
                  // the fraction shrinks it, while each weighted term here
                  // stays within the larger magnitude and the sum of the
                  // two is the finite value between the neighbours — the
                  // median's overflow rule, spelled for the general
                  // fraction.
                  (1.0 - frac) * lower + frac * higher
                }
              }
              Lower => lower
              Higher => higher
              Midpoint =>
                if lo_idx == hi_idx {
                  lower
                } else {
                  // The median's overflow rule: two near-`Double`-max
                  // neighbours overflow their sum before the halving, so halve
                  // first when it has — the true midpoint lies between them
                  // and is finite.
                  let sum = lower + higher
                  if sum.is_inf() {
                    lower / 2.0 + higher / 2.0
                  } else {
                    sum / 2.0
                  }
                }
            },
          )
        }
      }
    }
  }
}

///|
/// `arg_min` / `arg_max` reducer over an index window: the **window-local
/// position** (`k`, not the source row `at(k)`) of the first extremum cell,
/// as a `Scalar::Int` — so a group's arg is the offset within the group
/// and the whole-frame arg the row index. The order, the
/// null skip, and the `NaN` skip are `reduce_extremum`'s exactly (total over
/// every dtype; `Bool` `false < true`; `String` lexicographic), and a tie
/// keeps the **earliest** position — a deterministic choice. An empty /
/// all-null / all-`NaN` window reduces to `Scalar::Null`.
fn reduce_arg_extremum(
  storage : @column.ColumnStorage,
  want_min~ : Bool,
) -> (Int32, (Int32) -> Int32) -> @types.Scalar {
  let present = reduce_present(storage)
  match storage.data() {
    @column.ColumnData::I64(a) =>
      fn(n : Int32, at : (Int32) -> Int32) {
        arg_extremum_of(n, k => a[at(k)], k => !present(at(k)), (x, y) => {
          if want_min {
            x < y
          } else {
            x > y
          }
        })
      }
    @column.ColumnData::F64(a) =>
      fn(n : Int32, at : (Int32) -> Int32) {
        arg_extremum_of(
          n,
          k => a[at(k)],
          // `NaN` is treated as missing (skipped), as in `reduce_extremum`.
          k => !present(at(k)) || a[at(k)].is_nan(),
          (x, y) => if want_min { x < y } else { x > y },
        )
      }
    @column.ColumnData::Bool(a) =>
      fn(n : Int32, at : (Int32) -> Int32) {
        arg_extremum_of(
          n,
          k => a[at(k)],
          k => !present(at(k)),
          // `false < true`: under min `false` beats `true`, under max
          // reverse — `reduce_extremum`'s Bool order.
          (x, y) => if want_min { !x && y } else { x && !y },
        )
      }
    @column.ColumnData::Utf8(a) =>
      fn(n : Int32, at : (Int32) -> Int32) {
        arg_extremum_of(n, k => a.get(at(k)), k => !present(at(k)), (x, y) => {
          // `reduce_extremum`'s comparator (`compare_string_lex`), not the
          // native `String` order — the two disagree on length-mixed input
          // (the native order is shortlex), which would hand `arg_min` the
          // row of `max` and vice versa.
          if want_min {
            @text.compare_string_lex(x, y) < 0
          } else {
            @text.compare_string_lex(x, y) > 0
          }
        })
      }
  }
}

///|
/// The fold behind `reduce_arg_extremum`'s four backend arms: the local
/// position of the first cell the strict `better` prefers, `Scalar::Int(k)`
/// — or `Scalar::Null` when every position is skipped. `better` strict means
/// a tie never displaces the earlier position.
fn[T] arg_extremum_of(
  n : Int32,
  read : (Int32) -> T,
  skip : (Int32) -> Bool,
  better : (T, T) -> Bool,
) -> @types.Scalar {
  let mut best : T? = None
  let mut best_k = 0
  for k in 0..
          if better(v, b) {
            best = Some(v)
            best_k = k
          }
        None => {
          best = Some(v)
          best_k = k
        }
      }
    }
  }
  match best {
    Some(_) => @types.Scalar::Int(best_k.to_int64())
    None => @types.Scalar::Null
  }
}

///|
/// Count the distinct values in an index window — the total core
/// behind both the `n_unique` reduction and `Series::n_unique`, so the two
/// agree on what "distinct" means cell-for-cell. Cells are keyed by the same
/// `key_cell` normalisation `group_by` / `join` use, so "distinct value" means
/// exactly "distinct group": every `NaN` collapses into one bucket and `-0.0`
/// folds into `+0.0`. A null cell counts as one distinct value — the null
/// bucket, when the window has one, adds one to the keyed buckets. The
/// dtype dispatch resolves once into a per-row key
/// reader; only `Float` needs normalisation (which `key_cell` centralises),
/// the others key on the native value directly.
fn count_distinct(
  storage : @column.ColumnStorage,
) -> (Int32, (Int32) -> Int32) -> Int32 {
  let present = reduce_present(storage)
  let key_of : (Int32) -> 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 : Int32, at : (Int32) -> Int32) -> Int32 {
    let seen : Map[KeyCell, Unit] = Map([])
    let mut has_null = false
    for k in 0.. (Int32, (Int32) -> Int32) -> @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 : (Int32) -> @types.Scalar = i => @column.scalar_of(dt, data, i)
  fn(n : Int32, at : (Int32) -> Int32) {
    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
      }
    }
  }
}

///|
/// How `quantile` reads an answer that falls between two cells. Over the `m`
/// sorted non-null values, position `pos = q * (m - 1)` names the answer;
/// these say what to return when `pos` is not itself an index:
///
/// - `Nearest` (the default) — the value at `round(pos)`, halves rounding
///   up;
/// - `Linear` — the two neighbours weighted by `frac = pos - floor(pos)`
///   (numpy's default);
/// - `Lower` — the value at `floor(pos)`;
/// - `Higher` — the value at `ceil(pos)`;
/// - `Midpoint` — the mean of the `Lower` and `Higher` neighbours.
///
/// `quantile(0.5, Linear)` over an even count is the median's two-middle
/// average; `Nearest` / `Lower` / `Higher` always return a cell that is in
/// the column, `Linear` / `Midpoint` can manufacture a value between two.
pub(all) enum QuantileInterpolation {
  Nearest
  Linear
  Lower
  Higher
  Midpoint
} derive(Eq, Debug)

///|
pub extend QuantileInterpolation with Eq::{equal, not_equal}

///|
pub extend QuantileInterpolation with Debug::{to_repr}

///|
/// The fourteen 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
  Quantile(Double, QuantileInterpolation)
  NUnique
  ArgMin
  ArgMax
  Mode
  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` / `ArgMin` / `ArgMax` are always `Int`, `Mean` /
/// `Std` / `Var` / `Median` / `Quantile` 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` / `Quantile` `raise`
/// `TypeMismatch` here for a non-numeric column, before any window work
/// (`Quantile` also raising `InvalidOperation` for a `q` outside `[0, 1]`).
/// The reducer itself may refuse mid-fold: a decimal `sum` whose total
/// leaves the `Int64` range raises `InvalidOperation` from whichever
/// window crosses it (the exact dtype refuses instead of wrapping), so a
/// caller walks its windows in a `raise` context.
///
/// 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,
) -> (
  (Int32, (Int32) -> Int32) -> @types.Scalar raise @types.DataError,
  @types.DataType,
) raise @types.DataError {
  let storage = self.storage
  match op {
    Count => (lift_total(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 =>
      (lift_total(reduce_extremum(storage, want_min=true)), storage.dtype())
    Max =>
      (lift_total(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)
    Quantile(q, interpolation) =>
      (reduce_quantile(storage, q, interpolation), @types.DataType::Float)
    // `ArgMin` / `ArgMax` are the extremum reducers' positional twins: an
    // `Int` cell holding the **window-local position** of the extremum (its
    // `k`, so a group's arg is the offset within the group), total over
    // every dtype like `Min` / `Max`.
    ArgMin =>
      (
        lift_total(reduce_arg_extremum(storage, want_min=true)),
        @types.DataType::Int,
      )
    ArgMax =>
      (
        lift_total(reduce_arg_extremum(storage, want_min=false)),
        @types.DataType::Int,
      )
    // `mode` keeps the source dtype (the winning cell itself, whatever it
    // is).
    Mode => (lift_total(reduce_mode(storage)), storage.dtype())
    NUnique => {
      // The `n_unique` reducer over an index window: distinct value count
      // as `Scalar::Int`, total over every dtype and never null (an empty
      // window counts zero), over the shared `count_distinct` core
      // `Series::n_unique` also uses.
      let counter = count_distinct(storage)
      (
        fn(n : Int32, at : (Int32) -> Int32) {
          @types.Scalar::Int(counter(n, at).to_int64())
        },
        @types.DataType::Int,
      )
    }
    First =>
      (lift_total(reduce_first_last(storage, want_first=true)), storage.dtype())
    Last =>
      (
        lift_total(reduce_first_last(storage, want_first=false)),
        storage.dtype(),
      )
  }
}

///|
/// Adapt a total reducer to the raising reducer type `reducer_for` hands
/// out. The raising half exists for one refusal — the decimal `sum` past
/// the 18-digit magnitude — and the reducers that cannot refuse (a count,
/// an extremum, a first / last, a mode) keep their total signatures for
/// their direct callers, passing through here at the seam.
fn lift_total(
  reducer : (Int32, (Int32) -> Int32) -> @types.Scalar,
) -> (Int32, (Int32) -> Int32) -> @types.Scalar raise @types.DataError {
  fn(n, at) { reducer(n, at) }
}

///|
/// 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()
}

///|
/// `mode` reducer over an index window: the most frequent non-null cell,
/// as that cell's own `Scalar` — total over every dtype, keyed by the
/// shared `KeyCell` normalisation (`NaN` one bucket, `-0.0` folded). A tie
/// keeps the **first-appearing** value among the tied — one deterministic
/// single value per window, there being no list dtype to return every tied
/// value in. Nulls are skipped — the winning cell is a value; an empty /
/// all-null window reduces to `Scalar::Null`.
fn reduce_mode(
  storage : @column.ColumnStorage,
) -> (Int32, (Int32) -> Int32) -> @types.Scalar {
  let present = reduce_present(storage)
  let data = storage.data()
  let dtype = storage.dtype()
  // The every-dtype cell read: the column layer's `scalar_of` rider-aware
  // accessor, so a `Date` cell keys as its day count and reduces to a
  // `Date`.
  let cell_at = i => @column.scalar_of(dtype, data, i)
  fn(n : Int32, at : (Int32) -> Int32) {
    let count : Map[KeyCell, Int64] = Map([])
    let first_key : Map[KeyCell, Int32] = Map([])
    let mut best : KeyCell? = None
    let mut best_count : Int64 = 0
    let mut best_pos = 0
    for k in 0..
            first_key.get(key).unwrap_or(0) < first_key.get(b).unwrap_or(0)
          None => false
        }
        if c > best_count || (c == best_count && earlier_tie) {
          best = Some(key)
          best_count = c
          // The winner reads at its key's **first** appearance, not at
          // the occurrence that tipped the count: the `KeyCell`
          // normalisation folds spellings (`-0.0` into `+0.0`), so a
          // later occurrence of the same key would surface its spelling
          // — `-0.0` where the first-appearing value was `+0.0`.
          best_pos = first_key.get(key).unwrap_or(k)
        }
      }
    }
    match best {
      Some(_) => cell_at(at(best_pos))
      None => @types.Scalar::Null
    }
  }
}