// The CSV read path for `DataFrame`. It sits on top of NyaCSV for
// tokenisation and adds what a typeless format needs to reach a typed frame:
// header / no-header switching, per-column type inference
// (Int → Float → Bool → String), a configurable null-string mapping, ragged-row
// and quote strictness, and the projection / predicate seams the lazy scan
// pushes into. The write path is a file of its own (`io/csv_write.mbt`); what
// the two must agree on — which delimiters are legal — is decided here, by the
// validator both call.

///|
/// Options that control how a CSV input is parsed into a `DataFrame`.
///
/// - `has_header` — when `true`, the first row supplies column names.
///   When `false`, every row is treated as data and column names are
///   synthesised as `"column1"`, `"column2"`, … in declaration order; the
///   **first data row fixes the column count**, so a later wider row's
///   trailing cells are dropped under the lenient default (set
///   `strict_column_count = true` to reject a ragged row instead).
/// - `delimiter` — field separator passed through to NyaCSV.
/// - `infer_schema_rows` — number of leading rows scanned when guessing
///   each column's dtype. `0` (or any value `<= 0`) lifts the cap and
///   scans every row (Polars' `infer_schema_length=None`), trading a
///   slower inference pass for never mis-guessing a column from a prefix.
///   Cells past a finite window are still parsed under the chosen dtype;
///   one that doesn't fit is handled per `on_parse_error`.
/// - `null_values` — raw strings that should be treated as null cells
///   (both for type inference, which skips them, and for the final
///   typed column, where they become `None`). The default `[""]`
///   matches the empty string only, which is how an absent CSV cell is
///   typically represented. Read it back through the `null_values()`
///   accessor: the field itself is private, and both it and the
///   constructor copy, so the token list a reader uses cannot be changed
///   after the options are built.
/// - `strict_column_count` — when `true`, every data row must have
///   exactly as many cells as the header declares; a ragged row (too
///   few or too many cells) `raise ParseError` instead of being
///   silently null-padded (short row) or truncated (long row). Defaults
///   to `false`, preserving the lenient behaviour that tolerates ragged
///   rows (a permissive CSV idiom, and the reader's long-standing default).
/// - `on_parse_error` — what to do when a non-null cell past the
///   inference window fails to parse under its column's locked-in dtype.
///   `Raise` (default) fails with `ParseError(Cell(...))`; `Null`
///   downgrades the offending cell to a null cell and keeps the column's
///   inferred dtype (Polars' `ignore_errors=True`). See `OnParseError`.
/// - `allow_nonfinite_floats` — when `true` (default), the float probe
///   accepts every non-finite result: the `nan` / `inf` / `infinity` literals,
///   and a finite literal that overflows `Double` to a signed `Infinity`
///   (`1e999`), so a column of those tokens infers as `Float`. When
///   `false`, the probe rejects them, so such a column falls back to
///   `String` instead of being silently retyped to `Float` (a
///   non-finite token past the window is then a parse failure handled per
///   `on_parse_error`).
/// - `strict_quotes` — when `true`, the input is pre-scanned and an
///   unterminated quoted field, text after a closing quote, or a bare quote
///   inside an unquoted field `raise ParseError` instead of being repaired
///   by the lenient tokeniser. Defaults to `false`.
pub struct CsvReadOptions {
  has_header : Bool
  delimiter : Char
  infer_schema_rows : Int
  priv null_values : Array[String]
  strict_column_count : Bool
  on_parse_error : OnParseError
  allow_nonfinite_floats : Bool
  strict_quotes : Bool
}

///|
/// The raw strings treated as null cells. A fresh array is returned, so a
/// caller cannot reach into the options and change what the reader nulls out
/// — the same reason the constructor copies its input.
pub fn CsvReadOptions::null_values(self : CsvReadOptions) -> Array[String] {
  self.null_values.copy()
}

///|
/// Build read options. Every field has a default: header on, comma
/// delimiter, scan the first 100 rows for inference, treat the empty string
/// as null, tolerate ragged rows, fail the read on a cell that doesn't fit
/// its inferred dtype, accept non-finite float literals, and tokenise
/// leniently. `CsvReadOptions::CsvReadOptions()` is the all-defaults reader; name only what
/// differs, as in `CsvReadOptions::CsvReadOptions(delimiter=';', strict_quotes=true)`.
///
/// `null_values` is copied, so mutating the array afterwards cannot alter
/// the options.
pub fn CsvReadOptions::CsvReadOptions(
  has_header? : Bool = true,
  delimiter? : Char = ',',
  infer_schema_rows? : Int = DEFAULT_INFER_SCHEMA_ROWS,
  null_values? : Array[String] = [""],
  strict_column_count? : Bool = false,
  on_parse_error? : OnParseError = Raise,
  allow_nonfinite_floats? : Bool = true,
  strict_quotes? : Bool = false,
) -> CsvReadOptions {
  {
    has_header,
    delimiter,
    infer_schema_rows,
    null_values: null_values.copy(),
    strict_column_count,
    on_parse_error,
    allow_nonfinite_floats,
    strict_quotes,
  }
}

///|
/// Parse a CSV-encoded string into a `DataFrame`. The pipeline is:
///   1. NyaCSV tokenises the text into headers + rows. NyaCSV always
///      treats the first row as a header, so the `has_header = false`
///      case re-folds that row into the data section and synthesises
///      `column1`, `column2`, … names.
///   2. Per-column type inference walks the first `infer_schema_rows`
///      rows in order `Int → Float → Bool → String` (a null cell within
///      them is skipped but still counts toward the window). The
///      first option that accepts every probed cell wins; a column
///      with no non-null probes lands on `String` (no information to
///      disambiguate).
///   3. Null mapping replaces any raw cell whose verbatim string sits
///      in `null_values` with `None`. The mapping runs after
///      tokenisation, so quoted empty cells (`""`) and bare empty
///      cells are both honoured.
///   4. Each column is assembled into a typed `Series` and the result
///      is wrapped through `DataFrame::from_parts` (an empty input
///      short-circuits to `DataFrame::DataFrame([])` before this point).
///
/// Raises:
/// - `DuplicateColumn(name)` — two headers share a name.
/// - `ParseError(Message(...))` — when `strict_quotes` is `true`, the input contains
///   an unterminated quoted field, text after a closing quote, or a quote
///   inside an unquoted field; or — when `options.strict_column_count` is
///   `true` — a data row's cell count differs from the header's (reported as
///   `row N: expected C columns, got G`, 1-based over the data rows).
/// - `ParseError(Cell(...))` — a non-null cell does not parse under the
///   inferred dtype for its column (typically because a sample
///   row outside the inference window is malformed for the inferred
///   type), unless `options.on_parse_error` is `Null`, which downgrades
///   that cell to a null instead of failing the read.
/// - Whatever `DataFrame::from_parts` propagates — `DuplicateColumn` /
///   `LengthMismatch` / a negative-height `InvalidOperation`, none of which
///   can actually fire here (uniqueness is checked above; every column
///   is built off the same non-negative `rows.length()`).
///
/// By default, tokenisation keeps NyaCSV's lenient quote handling. An
/// unterminated quoted field at end-of-input is read as a complete field
/// (`1,"unclosed` → last cell `unclosed`); text after a closing quote is
/// folded into the same field (`"ab"cd` → `abcd`); and a lone quote inside
/// an unquoted field is kept literally (`ab"cd` → `ab"cd`). A strict
/// RFC-4180 reader rejects all three; `CsvReadOptions::CsvReadOptions(strict_quotes=true)`
/// runs a linear validation pass before tokenisation and rejects them as
/// `ParseError`.
pub fn parse_csv_str(
  content : String,
  options? : CsvReadOptions = CsvReadOptions::CsvReadOptions(),
) -> @frame.DataFrame raise @types.DataError {
  parse_csv_str_projected(content, options, None)
}

///|
/// Shared core behind `parse_csv_str` and `read_csv_projected`: the full
/// parse pipeline documented on `parse_csv_str`, with one extra knob —
/// `projection`.
///
/// - `None` — build every column (the eager `read_csv` / `parse_csv_str`
///   behaviour).
/// - `Some(names)` — build only the columns whose header name appears in
///   `names`, in the file's own header order, leaving the rest uninferred,
///   unparsed, and unmaterialised. This is the projection-pushdown path the
///   lazy `scan_csv` source uses: the optimizer narrows the read to exactly
///   the columns a pipeline consumes. A requested name that matches no header
///   is silently ignored (lenient projection, as in Polars): the
///   engine-internal `read_csv_projected` — an `#internal` seam, absent from
///   the generated interface — does not raise `ColumnNotFound` for an unknown
///   column, it just yields the ones that match. The lazy path only ever
///   requests real columns, and a stray `col("missing")` is still caught
///   downstream by `select`. A projection matching **no** header builds no
///   column at all and yields the `N×0` frame — the file's rows are still
///   counted, so a downstream `head` / `slice` bounds-checks against the real
///   height and the surfaced error stays the as-built plan's `ColumnNotFound`.
///
/// Every whole-input check runs *before* the projection filter and so is
/// unaffected by it — strict quote validation, the empty-input short-circuit,
/// duplicate-header rejection, and the optional ragged-row guard all see the
/// full input, so a projected read fails on malformed quoting, a malformed
/// header, or a ragged row exactly as a full read would. The single behavioural
/// difference is the
/// deliberate point of the feature: a non-null cell that fails to parse under
/// its column's inferred dtype only surfaces (`on_parse_error = Raise`) for a
/// column that is actually built, so a parse error confined to a *dropped*
/// column is not raised — that column is never parsed.
fn parse_csv_str_projected(
  content : String,
  options : CsvReadOptions,
  projection : Array[String]?,
  pruner? : (
    Array[String],
    (@frame.DataFrame) -> Array[Int] raise @types.DataError,
  ),
) -> @frame.DataFrame raise @types.DataError {
  // Refuse a delimiter that collides with the quote character or a row
  // terminator before parsing — the same configurations `format_csv` rejects,
  // so a value the writer can't emit unambiguously can't be read back either.
  validate_csv_delimiter(options.delimiter)
  // Strip a leading UTF-8 BOM (U+FEFF) before anything else, so a BOM-only file
  // still reads as empty. Shared with the JSON / NDJSON readers via
  // `strip_utf8_bom` so the formats can't drift apart.
  let content = strip_utf8_bom(content)
  // Short-circuit empty input. NyaCSV would otherwise produce a CSV
  // with no headers and no rows, which would collapse to an empty
  // 0×0 frame — handled the same way here, just without the parse.
  if content.is_empty() {
    return @frame.DataFrame::DataFrame([])
  }
  if options.strict_quotes {
    validate_csv_quotes(content, options.delimiter)
  }
  let csv_options : @nyacsv.CSVOptions = {
    delimiter: options.delimiter,
    allow_newlines_in_quotes: true,
    quote_char: '"',
    skip_empty_lines: true,
    trim_spaces: false,
  }
  let csv = @nyacsv.CSV::parse_string(content, options=csv_options)
  let (headers, rows) = split_csv_records(csv, options)
  let ncols = headers.length()
  if ncols == 0 {
    return @frame.DataFrame::DataFrame([])
  }
  validate_csv_shape(headers, rows, options)
  // Infer one dtype per column, then build the column — skipping any the
  // projection drops. `build_column` raises `ParseError(Cell(...))` for a cell
  // that doesn't fit the inferred dtype. The surviving columns keep the file's own
  // header order (a filtered projection, never a reorder), so a projected read
  // is column-for-column a sub-frame of the full read.
  // Build the null-token test once: an O(1) `Map` membership check shared by
  // inference and the typed build, so a large `null_values` list does not turn
  // every cell scan into an O(k) linear probe.
  let null_set : Map[String, Unit] = Map([])
  for v in options.null_values {
    null_set[v] = ()
  }
  let is_null = fn(v : String) -> Bool { null_set.contains(v) }
  let dtype_of = fn(col : Int) -> InferredDtype {
    // Inference always walks the *whole* file, even when a predicate will drop
    // rows, so a pruned read locks in the same dtypes an eager
    // `read_csv(..).filter(..)` would.
    infer_column_dtype(
      rows,
      col,
      options.infer_schema_rows,
      is_null,
      options.allow_nonfinite_floats,
    )
  }
  let build = fn(
    name : String,
    dt : InferredDtype,
    from : Array[Array[String]],
    col : Int,
    row_numbers : Array[Int]?,
  ) -> @series.Series raise @types.DataError {
    build_column(
      name,
      dt,
      from,
      col,
      is_null,
      options.on_parse_error,
      options.allow_nonfinite_floats,
      row_numbers,
    )
  }
  // Build one column over the records `keep` selects (`None` = every row).
  // `io` sees column names and a row-selection callback, never the expression
  // language; a named column the file lacks simply isn't built, so a pruner's
  // callback raises `ColumnNotFound` exactly as the eager filter would. `keep`
  // doubles as the survivor→original-row map, so a parse error in a kept row
  // reports its true file row (see `build_column`'s `file_row`). Inference runs
  // over the whole file either way, so a pruned read locks in the same dtypes.
  let build_over = fn(
    name : String,
    col : Int,
    keep : Array[Int]?,
  ) -> @series.Series raise @types.DataError {
    let records = match keep {
      None => rows
      Some(k) => Array::makei(k.length(), i => rows[k[i]])
    }
    build(name, dtype_of(col), records, col, keep)
  }
  assemble_pruned_frame(headers, rows.length(), projection, pruner, build_over)
}

///|
/// Settle the headers / rows split of a parsed CSV based on `has_header`.
/// NyaCSV always lifts the first record into `header()`, so the
/// `has_header = false` branch puts it back at the head of the data section
/// and synthesises `column1..columnN` names.
///
/// Both arms build fresh vectors rather than handing NyaCSV's own containers
/// on, so the caller owns what it gets.
fn split_csv_records(
  csv : @nyacsv.CSV,
  options : CsvReadOptions,
) -> (Array[String], Array[Array[String]]) {
  if options.has_header {
    return ([..csv.header()], [..csv.data()])
  }
  // An empty `header()` means the file held no records at all: there is no
  // first row to give back, and so no width to name columns from.
  let first = csv.header()
  let rows : Array[Array[String]] = [
    ..if !first.is_empty() {
      [first]
    },
    ..csv.data(),
  ]
  let ncols = if rows.is_empty() { 0 } else { rows[0].length() }
  ([ for i in 0.. "column\{i + 1}" ], rows)
}

///|
/// Whole-shape validation ahead of any per-cell work: reject a duplicate
/// header up front (`DataFrame::DataFrame` would catch it via `Schema::Schema`, but
/// surfacing it here spares a wasted inference pass), and — when
/// `strict_column_count` is on — `raise ParseError` for a data row whose
/// cell count differs from the header's, rather than silently null-padding
/// (short row) or truncating (long row) in the lenient `cell_value` path.
/// Checked after the header settles, so a malformed header surfaces first;
/// the 1-based index is over the data rows (the header is not counted).
fn validate_csv_shape(
  headers : Array[String],
  rows : Array[Array[String]],
  options : CsvReadOptions,
) -> Unit raise @types.DataError {
  let seen : Map[String, Unit] = Map([])
  for name in headers {
    if seen.contains(name) {
      raise @types.DataError::DuplicateColumn(name)
    }
    seen[name] = ()
  }
  if options.strict_column_count {
    let ncols = headers.length()
    for i, row in rows {
      if row.length() != ncols {
        raise @types.DataError::ParseError(
          @types.ParseErrorDetail::Message(
            "row \{i + 1}: expected \{ncols} columns, got \{row.length()}",
          ),
        )
      }
    }
  }
}

///|
/// Read a CSV file. `options` defaults to `CsvReadOptions::CsvReadOptions()`, so
/// `read_csv(path)` is the default-options read. `IOError` from the file
/// system surfaces as `raise IoError(message)`.
/// `CsvReadOptions::CsvReadOptions(strict_quotes=true)` rejects malformed quoting before
/// NyaCSV tokenisation, as documented by `parse_csv_str`.
pub fn read_csv(
  path : String,
  options? : CsvReadOptions = CsvReadOptions::CsvReadOptions(),
) -> @frame.DataFrame raise @types.DataError {
  let content = read_text(path)
  parse_csv_str(content, options~)
}

///|
/// Read a CSV file, keeping only the rows `predicate` selects — the file-backed
/// entry behind the lazy `scan_csv` source's **predicate push-down**. An
/// `#internal` engine seam, like `read_csv_projected`.
///
/// The read runs in two phases over one tokenisation of the whole file:
///   1. build only the `key_columns`, infer their dtypes from the whole file
///      exactly as an unfiltered read would, and ask `keep` — a callback the
///      lazy layer builds from the absorbed predicate — which rows survive;
///   2. build the `projection` columns for **those rows alone**.
///
/// Taking names and a callback rather than an expression keeps `io` below the
/// expression language: the reader knows *which* columns to resolve and which
/// rows survived, never *why*.
///
/// Dtypes therefore match an eager `read_csv(...).filter(...)` cell for cell —
/// inference always sees the full file — while the typed parse of a dropped
/// row's cells never runs. That is the deliberate consequence, and the same
/// one projection push-down already documents: a `ParseError` confined to a
/// row the predicate drops (in a column the predicate does not read) no longer
/// surfaces. Whole-input checks — strict quotes, a malformed header, a ragged
/// row — run before either phase and are unaffected.
#doc(hidden)
#internal(engine, "MoonFrame execution engine API")
pub fn read_csv_pruned(
  path : String,
  options : CsvReadOptions,
  projection : Array[String]?,
  key_columns : Array[String],
  keep : (@frame.DataFrame) -> Array[Int] raise @types.DataError,
) -> @frame.DataFrame raise @types.DataError {
  let content = read_text(path)
  parse_csv_str_projected(
    content,
    options,
    projection,
    pruner=(key_columns, keep),
  )
}

///|
/// Read a CSV file, building only the columns named in `projection` — the
/// file-backed entry behind the lazy `scan_csv` source's projection
/// pushdown. Identical to `read_csv` except that a column whose
/// header name is not in `projection` is never inferred, parsed, or
/// materialised (see `parse_csv_str_projected` for the precise contract,
/// including the one behavioural consequence: a parse error confined to a
/// dropped column does not surface). `IOError` from the file system surfaces
/// as `raise IoError(message)`. `CsvReadOptions::CsvReadOptions(strict_quotes=true)` validates the whole input
/// before applying the projection.
#doc(hidden)
#internal(engine, "MoonFrame execution engine API")
pub fn read_csv_projected(
  path : String,
  options : CsvReadOptions,
  projection : Array[String],
) -> @frame.DataFrame raise @types.DataError {
  let content = read_text(path)
  parse_csv_str_projected(content, options, Some(projection))
}

///|
/// State of the strict CSV quote validator. The validator mirrors RFC-4180's
/// field grammar without tokenising or allocating cells; NyaCSV remains the
/// parser after this pass succeeds.
priv enum CsvQuoteState {
  CsvFieldStart
  CsvUnquoted
  CsvQuoted
  CsvAfterQuote
}

///|
/// Reject the three malformed quote shapes NyaCSV otherwise accepts:
/// an opening quote without a close, non-separator text after a close, and a
/// quote inside an unquoted field. A doubled quote while quoted is an escape;
/// CR / LF are record separators outside quotes and ordinary content inside.
fn validate_csv_quotes(
  content : String,
  delimiter : Char,
) -> Unit raise @types.DataError {
  let mut state = CsvFieldStart
  for ch in content {
    state = match state {
      CsvFieldStart =>
        if ch == '"' {
          CsvQuoted
        } else if ch == delimiter || ch == '\r' || ch == '\n' {
          CsvFieldStart
        } else {
          CsvUnquoted
        }
      CsvUnquoted =>
        if ch == '"' {
          raise @types.DataError::ParseError(
            @types.ParseErrorDetail::Message(
              "malformed CSV quote: quote inside an unquoted field",
            ),
          )
        } else if ch == delimiter || ch == '\r' || ch == '\n' {
          CsvFieldStart
        } else {
          CsvUnquoted
        }
      CsvQuoted => if ch == '"' { CsvAfterQuote } else { CsvQuoted }
      CsvAfterQuote =>
        if ch == '"' {
          CsvQuoted
        } else if ch == delimiter || ch == '\r' || ch == '\n' {
          CsvFieldStart
        } else {
          raise @types.DataError::ParseError(
            @types.ParseErrorDetail::Message(
              "malformed CSV quote: unexpected character after a closing quote",
            ),
          )
        }
    }
  }
  if state is CsvQuoted {
    raise @types.DataError::ParseError(
      @types.ParseErrorDetail::Message(
        "malformed CSV quote: unterminated quoted field",
      ),
    )
  }
}

///|
/// Reject a delimiter that can't unambiguously frame fields. The quote
/// character is hard-coded as `'"'` and rows end at `'\n'` / `'\r'`, so a
/// delimiter equal to any of those collapses the field / record / quote
/// structure into one byte — a write emits unparseable output and a read splits
/// on the wrong boundary. A supplementary-plane (non-BMP) delimiter is rejected
/// too: the tokenizer matches one UTF-16 code unit at a time, so a delimiter
/// stored as a surrogate pair is never recognised and every cell would merge
/// into one column on a round-trip. Shared by `format_csv` and the reader so
/// both refuse the same configurations up front. (pandas / polars leave this as
/// GIGO; MoonFrame rejects it rather than corrupt a round-trip silently.)
fn validate_csv_delimiter(delimiter : Char) -> Unit raise @types.DataError {
  if delimiter == '"' || delimiter == '\n' || delimiter == '\r' {
    raise @types.DataError::InvalidOperation(
      "CSV delimiter must not be a double quote or a line terminator (\\n / \\r)",
    )
  }
  // `Char::to_int` is the Unicode code point; > 0xFFFF is the supplementary
  // plane, which UTF-16 stores as a surrogate pair the per-code-unit tokenizer
  // can never match.
  if delimiter.to_int() > 0xFFFF {
    raise @types.DataError::InvalidOperation(
      "CSV delimiter must be a single UTF-16 code unit (a Basic-Multilingual-Plane character)",
    )
  }
  // U+FEFF can never frame the first field unambiguously: every read strips
  // a leading U+FEFF as a byte-order mark before tokenizing, so a file whose
  // first field is empty loses its leading delimiter on read-back and every
  // cell of that record shifts left — the silent-corruption class this
  // validation exists to reject.
  if delimiter == '\u{FEFF}' {
    raise @types.DataError::InvalidOperation(
      "CSV delimiter must not be U+FEFF (stripped as a byte-order mark on read)",
    )
  }
}

///|
/// Read the cell at `(row_idx, col)` from a NyaCSV-produced data
/// matrix. Short rows (fewer cells than the header declares) fall
/// through to the empty string, which the null-mapping then turns
/// into `None` when the empty string is in `null_values`.
fn cell_value(rows : Array[Array[String]], row_idx : Int, col : Int) -> String {
  let row = rows[row_idx]
  if col < row.length() {
    row[col]
  } else {
    ""
  }
}

///|
/// Probe the first `limit` rows of `col` to pick a dtype (a null cell is
/// skipped but still counts toward the window). `Int → Float → Bool →
/// String` in order; whichever option accepts every probed cell wins. A
/// column whose probe window is entirely null defaults to `String`. Adapts
/// the CSV cell source (`String` cells from the row matrix) onto the shared
/// `infer_dtype` core.
fn infer_column_dtype(
  rows : Array[Array[String]],
  col : Int,
  limit : Int,
  is_null : (String) -> Bool,
  allow_nonfinite_floats : Bool,
) -> InferredDtype {
  infer_dtype(
    rows.length(),
    limit~,
    cell=i => cell_value(rows, i, col),
    is_null~,
    is_int=v => parse_int_opt(v) is Some(_),
    is_float=v => parse_float_opt(v, allow_nonfinite_floats) is Some(_),
    is_bool=v => parse_bool_opt(v) is Some(_),
  )
}

///|
/// `Some(value)` when `s` is a plain base-10 integer literal that fits in
/// a 64-bit `Int64`; `None` otherwise.
///
/// `@text.parse_decimal_int_opt` pre-screens for a plain decimal literal
/// because `@string.parse_int64` defaults to `base=0`, which would otherwise
/// accept `0x` / `0o` / `0b` prefixes (`0xFF` → 255) and underscore grouping
/// (`1_000` → 1000). Those forms are kept as strings by pandas / polars,
/// so accepting them during inference would silently retype columns of
/// hex codes, IDs, or formatted numbers. Restricting to decimal keeps
/// inference predictable. The same routine backs the `internal/column`
/// String→`Int` cast, so inference and an explicit cast agree.
fn parse_int_opt(s : String) -> Int64? {
  @text.parse_decimal_int_opt(s)
}

///|
/// `Some(value)` when `s` reads as a `Double` through
/// `@text.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 CSV numbers are written, and pandas / polars keep such cells as
/// strings.
///
/// `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
  }
}

///|
/// `Some(true)` / `Some(false)` for the literal strings `true` /
/// `false` in any ASCII casing; `None` otherwise. Numeric `0` / `1`
/// are intentionally **not** accepted — those round-trip as `Int`,
/// which the inference order picks up first.
fn parse_bool_opt(s : String) -> Bool? {
  if s.equal_ignore_ascii_case("true") {
    Some(true)
  } else if s.equal_ignore_ascii_case("false") {
    Some(false)
  } else {
    None
  }
}

///|
/// Materialise one column from the raw row matrix using the dtype
/// chosen by inference. The same null-mapping rules used by inference
/// run here; a non-null cell that fails to parse (the type was already
/// locked in by the inference pass) is handled per `on_parse_error` —
/// `Raise` surfaces `ParseError(Cell(...))`; `Null` makes it a null cell.
/// `allow_nonfinite_floats` matches the inference probe so a `Float`
/// column accepts (or rejects) `nan` / `inf` consistently on both passes.
fn build_column(
  name : String,
  dtype : InferredDtype,
  rows : Array[Array[String]],
  col : Int,
  isnull : (String) -> Bool,
  on_parse_error : OnParseError,
  allow_nonfinite_floats : Bool,
  row_numbers : Array[Int]?,
) -> @series.Series raise @types.DataError {
  let n = rows.length()
  let cell = fn(i : Int) -> String { cell_value(rows, i, col) }
  // A parse error must name the cell's original file row. `rows` may be the
  // compacted survivor list of a predicate-pruned read, so map the build index
  // back through `row_numbers` (identity when `None`, i.e. a full read); the
  // `+ 1` makes it the 1-based data row the eager reader reports. The NDJSON
  // reader threads the same original-position mapping (`kept_lines`).
  let file_row = fn(i : Int) -> Int {
    match row_numbers {
      None => i + 1
      Some(m) => m[i] + 1
    }
  }
  build_inferred_column(
    name,
    dtype,
    n,
    cell~,
    is_null=isnull,
    parse_int=parse_int_opt,
    parse_float=v => parse_float_opt(v, allow_nonfinite_floats),
    parse_bool=parse_bool_opt,
    render=v => v,
    locate=(dt, i, v) => {
      @types.DataError::ParseError(
        @types.ParseErrorDetail::Cell(
          @types.CellParseLocation::Row,
          name,
          file_row(i),
          dt,
          v,
        ),
      )
    },
    on_parse_error~,
  )
}