// Shared per-column dtype inference and typed-column construction for the
// CSV and JSON readers. Both readers walk an abstract "cell source" (a CSV
// row matrix indexed by column, or a JSON record array indexed by key) and
// pick the narrowest dtype that accepts every non-null probe, in the same
// priority order Int → Float → Bool → String. Factoring the skeleton here
// keeps that order — the windowed-inference rule (with `infer_schema_rows
// <= 0` meaning "scan every row") and
// the build-time parse-failure policy (`OnParseError`: raise vs. null the
// offending cell) — defined in exactly one place; each reader supplies only
// its own cell accessor, null test, and error wording. The numeric-text
// probes (`parse_int_opt` / `parse_float_opt`) are shared here too: the CSV
// inference path and the JSON readers' `repr`-digit recovery both route
// through them (the CSV-only `bool` probe stays with the CSV reader).
// The frame-level orchestration around these cores — survivor compaction
// and the two-phase pruned assembly — belongs to the pruned-read assembly.
//
// One-file-one-test exemption: this shared inference skeleton has no
// dedicated test file of its own. (doc-guard: unresolved) It is covered
// through the CSV / JSON / NDJSON reader blackbox tests that drive it.

///|
const DEFAULT_INFER_SCHEMA_ROWS : Int32 = 100

///|
/// The four dtypes inference can pick. `Null` is intentionally absent —
/// inference never produces it, so the builder match stays exhaustive
/// without an unreachable arm.
priv enum InferredDtype {
  InferInt
  InferFloat
  InferBool
  InferString
}

///|
/// A fresh copy of a caller's `dtypes` override map, so options built from
/// it cannot observe a later mutation of the caller's `Map` — the same
/// boundary the `null_values` copy draws.
fn copy_dtypes(
  source : Map[String, @types.DataType],
) -> Map[String, @types.DataType] {
  let out : Map[String, @types.DataType] = Map([])
  for name, dt in source {
    out[name] = dt
  }
  out
}

///|
/// Decode a `dtypes` override into the dtype the column build locks in.
/// The four dtypes inference can pick are exactly the four a reader can
/// parse from cell text; everything else a `DataType` can name — `Null`,
/// and the rider dtypes `Date` / `Timestamp` / `Decimal` — has no read
/// parser, so pinning one is refused loudly rather than silently ignored.
/// (Inference never produces these either: a written-back `Date` /
/// `Timestamp` / `Decimal` column reads back as `String` / `Float`.)
fn inferred_from_override(
  column : String,
  dt : @types.DataType,
) -> InferredDtype raise @types.DataError {
  match dt {
    Int => InferInt
    Float => InferFloat
    Bool => InferBool
    String => InferString
    Null | Date | Timestamp | Decimal(_) =>
      raise @types.DataError::InvalidOperation(
        "dtypes override: column \"\{column}\" cannot be pinned to \{dt}; only Int / Float / Bool / String have a read parser",
      )
  }
}

///|
/// Refuse a `dtypes` override naming a column the input does not carry —
/// a pinned name is the caller's assertion about the input, so a misspelt
/// one fails the read instead of being silently ignored (the opposite of
/// a projection's lenient unknown-name rule, which the optimizer's own
/// column list rides). Runs once per read, over the full header set, so
/// it holds for a projected read too.
fn check_override_columns(
  dtypes : Map[String, @types.DataType],
  headers : Array[String],
) -> Unit raise @types.DataError {
  for name, _ in dtypes {
    if !headers.contains(name) {
      raise @types.DataError::InvalidOperation(
        "dtypes override names column \"\{name}\", which the input does not carry",
      )
    }
  }
}

///|
/// What a reader does when a non-null cell fails to parse under the dtype
/// that inference locked in for its column — in practice a value past the
/// inference window that doesn't fit (e.g. a `String` in a column inferred
/// `Int` from its first rows).
///
/// - `Raise` (every reader's default) — surface `ParseError(Cell(...))`,
///   failing the whole read. The strict, lossless behaviour: a malformed
///   cell is never silently dropped.
/// - `Null` — downgrade the offending cell to a null cell and keep going,
///   the resilient behaviour. The column keeps its inferred dtype; only
///   the unparseable cells become null.
///
/// Shared by `CsvReadOptions` and `JsonReadOptions` and
/// consumed in `build_typed_column`'s parse-failure branch, so the three
/// readers agree on the policy.
pub(all) enum OnParseError {
  Raise
  Null
} derive(Eq, Debug)

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

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

///|
/// Pick the narrowest dtype that accepts every non-null cell in the first
/// `limit` rows, in priority order `Int → Float → Bool → String`. A
/// `limit <= 0` lifts the cap and scans every row, so a dtype that only
/// becomes clear deep in the data is inferred rather than guessed from a
/// prefix.
///
/// `cell(i)` reads the raw cell at row `i`; `is_null(c)` reports whether it
/// is a null cell (skipped by inference); `is_int` / `is_float` / `is_bool`
/// report whether a non-null cell parses as that dtype. A probe window
/// that is entirely null falls back to `String` (no signal to
/// disambiguate). A column that eliminates Int, Float, and Bool is
/// `String` whether it is pure-string or mixed, so once all three are out
/// we stop probing early — the remaining cells cannot change the outcome.
fn[Cell] infer_dtype(
  n : Int32,
  limit~ : Int32,
  cell~ : (Int32) -> Cell,
  is_null~ : (Cell) -> Bool,
  is_int~ : (Cell) -> Bool,
  is_float~ : (Cell) -> Bool,
  is_bool~ : (Cell) -> Bool,
) -> InferredDtype {
  let scan = if limit <= 0 || limit > n { n } else { limit }
  let mut all_int = true
  let mut all_float = true
  let mut all_bool = true
  let mut saw_any = false
  for i in 0.. Int64? {
  @text.parse_decimal_int_opt(s)
}

///|
/// `Some(value)` when `s` reads as a `Double` through
/// `parse_plain_double_opt`; `None` when it does not.
///
/// Strings containing `_` are rejected up front: `@string.parse_double`
/// accepts underscore grouping (`1_000` → `1000.0`), but — like the
/// `0x` / `0o` / `0b` forms screened out of `parse_int_opt` — that is not
/// how the text formats write numbers.
///
/// `allow_nonfinite` gates every non-finite result: the `nan` / `inf` /
/// `infinity` literals, and equally a finite literal whose magnitude overflows
/// `Double` (`1e999`), which IEEE 754 rounds to a signed `Infinity` rather than
/// failing. When `true` they are accepted and become
/// `Double` NaN / ±Infinity (the historical default); when `false` they
/// are rejected (`None`), so a column of such tokens stays `String` rather
/// than being silently retyped to `Float`. The same predicate gates both
/// the inference probe and the typed-column build, so the two agree on
/// whether a non-finite literal counts as a `Float`.
fn parse_float_opt(s : String, allow_nonfinite : Bool) -> Double? {
  match @text.parse_plain_double_opt(s) {
    Some(x) =>
      if !allow_nonfinite && (x.is_nan() || x.is_inf()) {
        None
      } else {
        Some(x)
      }
    None => None
  }
}

///|
/// Materialise one typed column from an abstract cell source. `cell(i)`
/// reads the raw cell, `is_null(c)` maps a null cell to `None`, `parse(c)`
/// converts a non-null cell to `Some(value)` (or `None` when it doesn't fit
/// the locked-in dtype), and `build` wraps the option array into a
/// `Series`. Used for the parseable dtypes (`Int` / `Float` / `Bool`); the
/// `String` fallback goes through the total `build_string_column` below (its
/// `render` never fails, so it needs no `parse` / `err` / `on_parse_error`).
///
/// A non-null cell that `parse` rejects is handled per `on_parse_error`:
/// `Raise` surfaces `err(i, c)` (the strict default, failing the read),
/// `Null` downgrades it to a null cell so the read continues with the
/// column's inferred dtype intact.
fn[Cell, T] build_typed_column(
  name : String,
  n : Int32,
  cell~ : (Int32) -> Cell,
  is_null~ : (Cell) -> Bool,
  parse~ : (Cell) -> T?,
  err~ : (Int32, Cell) -> @types.DataError,
  on_parse_error~ : OnParseError,
  build~ : (String, Array[T?]) -> @series.Series,
) -> @series.Series raise @types.DataError {
  let values : Array[T?] = Array::makei(n, i => {
    let c = cell(i)
    if is_null(c) {
      None
    } else {
      match parse(c) {
        Some(x) => Some(x)
        None =>
          match on_parse_error {
            Raise => raise err(i, c)
            Null => None
          }
      }
    }
  })
  build(name, values)
}

///|
/// Materialise a `String` column from an abstract cell source — the total
/// counterpart of `build_typed_column` for the `String` fallback dtype.
/// `cell(i)` reads the raw cell, `is_null(c)` maps a null cell to `None`, and
/// `render(c)` stringifies a non-null cell (it never fails, so there is no
/// `parse` / `err` / `on_parse_error`). The CSV reader renders with the
/// identity; the JSON reader with its value-to-string fallback.
fn[Cell] build_string_column(
  name : String,
  n : Int32,
  cell : (Int32) -> Cell,
  is_null : (Cell) -> Bool,
  render : (Cell) -> String,
) -> @series.Series {
  @series.Series::from_string_options(
    name,
    Array::makei(n, i => {
      let c = cell(i)
      if is_null(c) {
        None
      } else {
        Some(render(c))
      }
    }),
  )
}

///|
/// The four-arm `InferredDtype` dispatch shared by the CSV and JSON readers:
/// route the three parseable dtypes through `build_typed_column` (each with its
/// own parser and a `locate`d error) and the `String` fallback through the
/// total `build_string_column`. Each reader supplies only its cell accessor,
/// null test, per-dtype parsers, string render, and error locator —
/// `locate(dt, i, c)` builds the `ParseError` for the cell at index `i` that
/// failed to fit dtype `dt`, letting each format place the coordinate (a CSV
/// data row, an NDJSON line, a JSON record) and stringify the offending value
/// its own way. The match, and the `Int → Float → Bool → String` order it
/// mirrors from `infer_dtype`, live here once.
fn[Cell] build_inferred_column(
  name : String,
  dtype : InferredDtype,
  n : Int32,
  cell~ : (Int32) -> Cell,
  is_null~ : (Cell) -> Bool,
  parse_int~ : (Cell) -> Int64?,
  parse_float~ : (Cell) -> Double?,
  parse_bool~ : (Cell) -> Bool?,
  render~ : (Cell) -> String,
  locate~ : (@types.DataType, Int32, Cell) -> @types.DataError,
  on_parse_error~ : OnParseError,
) -> @series.Series raise @types.DataError {
  match dtype {
    InferInt =>
      build_typed_column(
        name,
        n,
        cell~,
        is_null~,
        parse=parse_int,
        err=(i, c) => locate(@types.DataType::Int, i, c),
        on_parse_error~,
        build=@series.Series::from_int_options,
      )
    InferFloat =>
      build_typed_column(
        name,
        n,
        cell~,
        is_null~,
        parse=parse_float,
        err=(i, c) => locate(@types.DataType::Float, i, c),
        on_parse_error~,
        build=@series.Series::from_float_options,
      )
    InferBool =>
      build_typed_column(
        name,
        n,
        cell~,
        is_null~,
        parse=parse_bool,
        err=(i, c) => locate(@types.DataType::Bool, i, c),
        on_parse_error~,
        build=@series.Series::from_bool_options,
      )
    InferString => build_string_column(name, n, cell, is_null, render)
  }
}