Skip to main content

gam_data/
lib.rs

1use csv::{ReaderBuilder, StringRecord};
2use ndarray::{Array2, ArrayViewMut1, Axis, s};
3use rayon::prelude::*;
4use serde::{Deserialize, Serialize};
5use std::cmp::Ordering;
6use std::collections::{HashMap, HashSet};
7use std::fmt;
8use std::path::Path;
9
10fn natural_level_cmp(a: &str, b: &str) -> Ordering {
11    let mut ia = 0;
12    let mut ib = 0;
13    let ba = a.as_bytes();
14    let bb = b.as_bytes();
15    while ia < ba.len() && ib < bb.len() {
16        if ba[ia].is_ascii_digit() && bb[ib].is_ascii_digit() {
17            let sa = ia;
18            let sb = ib;
19            while ia < ba.len() && ba[ia].is_ascii_digit() {
20                ia += 1;
21            }
22            while ib < bb.len() && bb[ib].is_ascii_digit() {
23                ib += 1;
24            }
25            let da = &a[sa..ia];
26            let db = &b[sb..ib];
27            let ta = da.trim_start_matches('0');
28            let tb = db.trim_start_matches('0');
29            let ta = if ta.is_empty() { "0" } else { ta };
30            let tb = if tb.is_empty() { "0" } else { tb };
31            match ta.len().cmp(&tb.len()).then_with(|| ta.cmp(tb)) {
32                Ordering::Equal if da.len() != db.len() => return da.len().cmp(&db.len()),
33                Ordering::Equal => {}
34                ord => return ord,
35            }
36        } else {
37            match ba[ia].cmp(&bb[ib]) {
38                Ordering::Equal => {
39                    ia += 1;
40                    ib += 1;
41                }
42                ord => return ord,
43            }
44        }
45    }
46    ba.len().cmp(&bb.len())
47}
48
49fn sort_levels_canonical(levels: &mut [String]) {
50    levels.sort_by(|a, b| natural_level_cmp(a, b));
51}
52
53/// Canonical bit key for a floating-point categorical / grouping level.
54///
55/// Factor dummies, random-effect groups, `by=` gates and factor-smooth blocks
56/// all identify a level by the raw bits of its numeric code — they intern the
57/// observed codes with `f64::to_bits()` and, at fit/predict time, gate each row
58/// by `data_bits == level_bits`. Raw `to_bits()` is a **bit** identity, not the
59/// **numeric** equality IEEE-754 defines, and the two disagree in exactly two
60/// places:
61///
62/// * **Signed zero.** `+0.0` is `0x0000_0000_0000_0000` and `-0.0` is
63///   `0x8000_0000_0000_0000`, yet IEEE-754 guarantees `+0.0 == -0.0`. Keying on
64///   raw bits splits one physical group into two: a row whose code is `-0.0`
65///   matches no `+0.0` dummy, so its factor / random effect silently drops and
66///   the prediction collapses onto the intercept / population mean. Signed zero
67///   arises routinely from ordinary float arithmetic on a computed group column
68///   (`-1.0 * 0.0`, a centred/differenced column landing on `-0.0`, `np.round`
69///   emitting `-0.0`), and the miss is silent — no schema error, `check()` still
70///   reports `ok=True`. See #2145 (random effect) and #2146 (factor dummy).
71/// * **NaN.** Every quiet/signalling NaN payload and sign bit denotes "not a
72///   number", so `2^53`-ish distinct bit patterns would otherwise intern as
73///   distinct levels. (NaN group codes are rejected upstream on most paths, but
74///   canonicalising here keeps the key numerically honest regardless.)
75///
76/// This maps both encodings to a single canonical key while leaving every
77/// ordinary finite value bit-stable, so that
78/// `canonical_level_bits(a) == canonical_level_bits(b)` iff `a` and `b` name the
79/// same real-valued level. Interning and lookup must **both** route through this
80/// function; because it is idempotent, applying it to an already-canonical
81/// frozen level set is a no-op.
82#[inline]
83pub fn canonical_level_bits(v: f64) -> u64 {
84    if v == 0.0 {
85        // Matches both +0.0 and -0.0 (IEEE-754: -0.0 == 0.0); collapse to +0.0.
86        0.0_f64.to_bits()
87    } else if v.is_nan() {
88        // Collapse every NaN payload/sign to one canonical quiet-NaN key.
89        f64::NAN.to_bits()
90    } else {
91        v.to_bits()
92    }
93}
94
95// ---------------------------------------------------------------------------
96// Typed error
97// ---------------------------------------------------------------------------
98
99/// Typed error variants for the data-loading module.
100///
101/// Public entry points continue to return `Result<_, String>`; this enum is
102/// materialized at leaf sites and converted at the boundary via
103/// `From<DataError> for String` so error text remains byte-identical to the
104/// previous ad-hoc `format!(...)` output.
105#[derive(Debug, Clone)]
106pub enum DataError {
107    /// Schema/column shape disagrees with the file: row width mismatch,
108    /// requested column missing from headers, schema-declared kind violated by
109    /// a row, or an unseen categorical level encountered under
110    /// `UnseenCategoryPolicy::Error`.
111    SchemaMismatch { reason: String },
112    /// Failed to open, decode, or read structural bytes of the source
113    /// (CSV/TSV row read, parquet metadata, file extension detection, Arrow
114    /// record-batch reads, or dictionary decoding).
115    ParseError { reason: String },
116    /// Internal encoding bookkeeping failed: a categorical map expected by the
117    /// schema path was missing, or a level expected to be present in the
118    /// per-column inference state was not found during fix-up.
119    EncodingFailure { reason: String },
120    /// The source has no headers, no rows, or contains an empty / missing
121    /// field at a row that requires a value.
122    EmptyInput { reason: String },
123    /// A cell value cannot be used as a feature: non-finite float, Arrow null,
124    /// or an unsupported Arrow data type for the column.
125    InvalidValue { reason: String },
126    /// A formula or call site references a column name that is not present in
127    /// the input data. Structured so the FFI boundary can raise a typed
128    /// Python exception (`gamfit.ColumnNotFoundError`) carrying the missing
129    /// name, available columns, and similarity suggestions as attributes —
130    /// not as a parsed-back-out substring of the human display text.
131    ///
132    ColumnNotFound {
133        /// The missing column name, exactly as the user wrote it.
134        name: String,
135        /// Optional role label (`"response"`, `"entry"`, `"exit"`, etc.)
136        /// supplied at the resolution site to disambiguate which slot in the
137        /// formula referenced the bad name. `None` for bare term references.
138        role: Option<String>,
139        /// All headers present in the input table at resolution time, sorted.
140        available: Vec<String>,
141        /// Cheap similarity suggestions (case-insensitive substring or
142        /// shared-prefix length ≥ 3), sorted; empty when no header is close.
143        similar: Vec<String>,
144        /// True iff the available set has exactly one entry and that entry
145        /// contains a literal tab — i.e. the user almost certainly handed gam
146        /// a TSV file under a `.csv` filename. Surfaced as a structured
147        /// boolean rather than re-parsed from prose at the boundary.
148        tsv_hint: bool,
149    },
150}
151
152impl DataError {
153    /// Build a typed `ColumnNotFound` from the column map of the resolved
154    /// dataset. Centralises the similarity / TSV-hint heuristics that the
155    /// legacy `missing_column_message` helper used to perform inline so all
156    /// callers — leaf `resolve_col*` shims and the multi-column requested-
157    /// columns aggregator — produce identical payloads.
158    pub fn column_not_found(
159        col_map: &HashMap<String, usize>,
160        name: &str,
161        role: Option<&str>,
162    ) -> Self {
163        let target_lower = name.to_lowercase();
164        let mut similar: Vec<String> = col_map
165            .keys()
166            .filter(|k| {
167                let k_lower = k.to_lowercase();
168                k_lower.contains(&target_lower)
169                    || target_lower.contains(&k_lower)
170                    || shared_prefix(&k_lower, &target_lower) >= 3
171            })
172            .cloned()
173            .collect();
174        similar.sort_unstable();
175        let mut available: Vec<String> = col_map.keys().cloned().collect();
176        available.sort_unstable();
177        let tsv_hint = available.len() == 1 && available[0].contains('\t');
178        Self::ColumnNotFound {
179            name: name.to_string(),
180            role: role.map(str::to_string),
181            available,
182            similar,
183            tsv_hint,
184        }
185    }
186}
187
188impl fmt::Display for DataError {
189    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
190        match self {
191            DataError::SchemaMismatch { reason }
192            | DataError::ParseError { reason }
193            | DataError::EncodingFailure { reason }
194            | DataError::EmptyInput { reason }
195            | DataError::InvalidValue { reason } => f.write_str(reason),
196            DataError::ColumnNotFound {
197                name,
198                role,
199                available,
200                similar,
201                tsv_hint,
202            } => {
203                let label = match role {
204                    Some(r) => format!("{r} column '{name}'"),
205                    None => format!("column '{name}'"),
206                };
207                let tsv_suffix = if *tsv_hint {
208                    " — your file appears to be tab-separated; gam expects comma-separated CSV. \
209         Replace tabs with commas, or pre-convert with `tr '\\t' ',' < file.tsv > file.csv`."
210                } else {
211                    ""
212                };
213                if similar.is_empty() {
214                    write!(
215                        f,
216                        "{label} not found in data. Available columns: [{}]{tsv_suffix}",
217                        available.join(", ")
218                    )
219                } else {
220                    write!(
221                        f,
222                        "{label} not found in data. Did you mean one of [{}]? Full list: [{}]{tsv_suffix}",
223                        similar.join(", "),
224                        available.join(", ")
225                    )
226                }
227            }
228        }
229    }
230}
231
232impl std::error::Error for DataError {}
233
234impl From<DataError> for String {
235    fn from(err: DataError) -> String {
236        err.to_string()
237    }
238}
239
240// ---------------------------------------------------------------------------
241// Public types
242// ---------------------------------------------------------------------------
243
244#[derive(Clone, Debug, Serialize, Deserialize)]
245pub struct DataSchema {
246    pub columns: Vec<SchemaColumn>,
247}
248
249#[derive(Clone, Debug, Serialize, Deserialize)]
250pub struct SchemaColumn {
251    pub name: String,
252    pub kind: ColumnKindTag,
253    #[serde(default)]
254    pub levels: Vec<String>,
255}
256
257#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
258#[serde(rename_all = "kebab-case")]
259pub enum ColumnKindTag {
260    Continuous,
261    Binary,
262    Categorical,
263}
264
265#[derive(Clone, Debug, Eq, PartialEq)]
266pub enum UnseenCategoryPolicy {
267    Error,
268    EncodeUnknownForColumns(HashSet<String>),
269}
270
271impl UnseenCategoryPolicy {
272    pub fn encode_unknown_for_columns(columns: HashSet<String>) -> Self {
273        if columns.is_empty() {
274            Self::Error
275        } else {
276            Self::EncodeUnknownForColumns(columns)
277        }
278    }
279
280    fn unseen_code_for(&self, column_name: &str, level_count: usize) -> Option<f64> {
281        match self {
282            Self::Error => None,
283            Self::EncodeUnknownForColumns(columns) => {
284                columns.contains(column_name).then_some(level_count as f64)
285            }
286        }
287    }
288}
289
290#[derive(Clone, Debug)]
291pub struct EncodedDataset {
292    pub headers: Vec<String>,
293    pub values: Array2<f64>,
294    pub schema: DataSchema,
295    pub column_kinds: Vec<ColumnKindTag>,
296}
297
298impl EncodedDataset {
299    pub fn column_map(&self) -> HashMap<String, usize> {
300        self.headers
301            .iter()
302            .enumerate()
303            .map(|(index, header)| (header.clone(), index))
304            .collect()
305    }
306
307    /// Per-column finite (min, max) of the training values, parallel to
308    /// `headers`. Columns with no finite values default to (0.0, 0.0) so that
309    /// downstream clipping is a no-op for them. Used to populate
310    /// `training_feature_ranges` so prediction can clip out-of-hull inputs
311    /// to the training bounding box.
312    pub fn feature_ranges(&self) -> Vec<(f64, f64)> {
313        // Iterate column-by-column (contiguous in C-order Array2 along axis 0
314        // only when the array is Fortran-order; here Array2 is row-major so
315        // each column is strided. However, scanning one column at a time keeps
316        // each column's working set hot, lets rayon parallelize across
317        // columns, and avoids the previous outer-col/inner-row pattern that
318        // re-streamed all rows per column with stride `p`.
319        self.values
320            .axis_iter(Axis(1))
321            .into_par_iter()
322            .map(|col| {
323                let (lo, hi) =
324                    col.iter()
325                        .fold((f64::INFINITY, f64::NEG_INFINITY), |(lo, hi), &v| {
326                            if v.is_finite() {
327                                (lo.min(v), hi.max(v))
328                            } else {
329                                (lo, hi)
330                            }
331                        });
332                if !lo.is_finite() || !hi.is_finite() {
333                    (0.0, 0.0)
334                } else {
335                    (lo, hi)
336                }
337            })
338            .collect()
339    }
340}
341
342fn shared_prefix(a: &str, b: &str) -> usize {
343    a.chars()
344        .zip(b.chars())
345        .take_while(|(ca, cb)| ca == cb)
346        .count()
347}
348
349// ---------------------------------------------------------------------------
350// Format detection
351// ---------------------------------------------------------------------------
352
353#[derive(Clone, Copy, Debug, Eq, PartialEq)]
354enum DataFormat {
355    Csv,
356    Tsv,
357    Parquet,
358}
359
360fn detect_format(path: &Path) -> Result<DataFormat, DataError> {
361    let ext = path
362        .extension()
363        .and_then(|s| s.to_str())
364        .unwrap_or_default()
365        .to_ascii_lowercase();
366    match ext.as_str() {
367        "csv" => Ok(DataFormat::Csv),
368        "tsv" | "txt" | "tab" => Ok(DataFormat::Tsv),
369        "parquet" | "pq" | "pqt" => Ok(DataFormat::Parquet),
370        other => Err(DataError::ParseError {
371            reason: format!(
372                "unsupported data file extension '.{other}'; expected csv, tsv, txt, parquet, or pq: '{}'",
373                path.display()
374            ),
375        }),
376    }
377}
378
379// ---------------------------------------------------------------------------
380// Unified public API  — format auto-detected, zero extra CLI args
381// ---------------------------------------------------------------------------
382
383pub fn load_dataset_projected(
384    path: &Path,
385    requested_columns: &[String],
386) -> Result<EncodedDataset, DataError> {
387    load_dataset_projected_with_categorical_roles(path, requested_columns, &HashSet::new())
388}
389
390/// Schema-inferring projected loader that forces a set of columns to
391/// [`ColumnKindTag::Categorical`] regardless of whether their labels parse as
392/// numbers.
393///
394/// An untyped CSV/TSV/parquet-numeric frame cannot carry the dtype the typed
395/// Python frame stamps via [`CATEGORICAL_CELL_SENTINEL`], so the value-based
396/// inferer would otherwise demote an integer/numeric-coded factor (e.g. a
397/// `group(region)` grouping coded `0,1,2,3`) to `Continuous` and fit a single
398/// numeric ramp instead of one centred factor level per code. That makes the
399/// CLI's design strictly lower-capacity than the Python `gamfit.fit` design for
400/// the same data, which generalizes worse on every seed.
401///
402/// `categorical_roles` is keyed on the *formula role*, not on a value
403/// heuristic: a column is forced categorical only when the formula uses it in a
404/// role that is a factor by construction (`group(g)` / `factor(g)` / `re(g)`
405/// random-effect terms, or a categorical/multinomial response). A bare `+ x`
406/// linear term and a smooth argument `s(x)` are deliberately NOT included — they
407/// stay value-inferred, so a genuinely continuous integer covariate like
408/// `s(age)` or `+ age` remains `Continuous`. This mirrors the Python sentinel
409/// outcome (`force_categorical`, the column-major inferer) while keying it on
410/// the role the user actually declared.
411pub fn load_dataset_projected_with_categorical_roles(
412    path: &Path,
413    requested_columns: &[String],
414    categorical_roles: &HashSet<&str>,
415) -> Result<EncodedDataset, DataError> {
416    match detect_format(path)? {
417        DataFormat::Csv => {
418            load_delimited_inferred(path, b',', requested_columns, categorical_roles)
419        }
420        DataFormat::Tsv => {
421            load_delimited_inferred(path, b'\t', requested_columns, categorical_roles)
422        }
423        DataFormat::Parquet => load_parquet_inferred(path, requested_columns, categorical_roles),
424    }
425}
426
427pub fn load_datasetwith_schema_projected(
428    path: &Path,
429    schema: &DataSchema,
430    unseen_policy: UnseenCategoryPolicy,
431    requested_columns: &[String],
432) -> Result<EncodedDataset, DataError> {
433    match detect_format(path)? {
434        DataFormat::Csv => {
435            load_delimited_with_schema(path, b',', schema, unseen_policy, requested_columns)
436        }
437        DataFormat::Tsv => {
438            load_delimited_with_schema(path, b'\t', schema, unseen_policy, requested_columns)
439        }
440        DataFormat::Parquet => {
441            load_parquet_with_schema(path, schema, unseen_policy, requested_columns)
442        }
443    }
444}
445
446// ---------------------------------------------------------------------------
447// CSV convenience loader — infers the schema from the file header.
448// ---------------------------------------------------------------------------
449
450pub fn load_csvwith_inferred_schema(path: &Path) -> Result<EncodedDataset, DataError> {
451    load_delimited_inferred(path, b',', &[], &HashSet::new())
452}
453
454// ---------------------------------------------------------------------------
455// Delimited (CSV / TSV) — streaming, columnar, single-pass
456// ---------------------------------------------------------------------------
457
458/// Maximum number of rows used for schema inference when no schema is provided.
459/// Prefix a typed Python frame stamps onto a cell that originates from a
460/// genuinely-categorical source column (string / object / categorical dtype).
461/// The column-major inference (`infer_and_encode_column_major`) and the
462/// schema-guided predict ingest (`gam-pyffi::string_records_from_rows`) both
463/// strip this prefix before recording or matching a level; its presence forces
464/// the column to `Categorical` even when every label parses as a number, so a
465/// string column labeled "0","1","2" is one centred factor level per label
466/// rather than a numeric ramp (#1317 / #1318). A leading NUL never appears in a
467/// numeric literal, so an untyped CSV/array frame (no prefix) is unaffected.
468pub const CATEGORICAL_CELL_SENTINEL: char = '\u{0}';
469
470/// Strip the leading [`CATEGORICAL_CELL_SENTINEL`] from a cell if present,
471/// returning the clean text and whether the marker was found.
472pub fn strip_categorical_sentinel(cell: &str) -> (&str, bool) {
473    match cell.strip_prefix(CATEGORICAL_CELL_SENTINEL) {
474        Some(rest) => (rest, true),
475        None => (cell, false),
476    }
477}
478
479fn resolve_requested_columns(
480    all_headers: &[String],
481    requested_columns: &[String],
482) -> Result<Vec<usize>, DataError> {
483    if requested_columns.is_empty() {
484        return Ok((0..all_headers.len()).collect());
485    }
486
487    let requested_set: HashSet<&str> = requested_columns.iter().map(String::as_str).collect();
488    let mut selected = Vec::with_capacity(requested_set.len());
489    for (idx, name) in all_headers.iter().enumerate() {
490        if requested_set.contains(name.as_str()) {
491            selected.push(idx);
492        }
493    }
494
495    if selected.len() != requested_set.len() {
496        let available_map: HashMap<String, usize> = all_headers
497            .iter()
498            .enumerate()
499            .map(|(index, header)| (header.clone(), index))
500            .collect();
501        let missing = requested_columns
502            .iter()
503            .filter(|name| !available_map.contains_key(name.as_str()))
504            .map(|name| {
505                DataError::column_not_found(&available_map, name, Some("requested")).to_string()
506            })
507            .collect::<Vec<_>>();
508        return Err(DataError::SchemaMismatch {
509            reason: missing.join("; "),
510        });
511    }
512
513    Ok(selected)
514}
515
516fn projected_headers(all_headers: &[String], selected_indices: &[usize]) -> Vec<String> {
517    selected_indices
518        .iter()
519        .map(|&idx| all_headers[idx].clone())
520        .collect()
521}
522
523fn load_delimited_inferred(
524    path: &Path,
525    delimiter: u8,
526    requested_columns: &[String],
527    categorical_roles: &HashSet<&str>,
528) -> Result<EncodedDataset, DataError> {
529    let t_open = std::time::Instant::now();
530    let mut rdr = ReaderBuilder::new()
531        .has_headers(true)
532        .delimiter(delimiter)
533        .from_path(path)
534        .map_err(|e| DataError::ParseError {
535            reason: format!("failed to open '{}': {e}", path.display()),
536        })?;
537
538    let all_headers: Vec<String> = rdr
539        .headers()
540        .map_err(|e| DataError::ParseError {
541            reason: format!("failed to read headers: {e}"),
542        })?
543        .iter()
544        .map(|s| s.trim().to_string())
545        .collect();
546    if all_headers.is_empty() {
547        return Err(DataError::EmptyInput {
548            reason: "file has no headers".to_string(),
549        });
550    }
551    let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
552    let headers = projected_headers(&all_headers, &selected_indices);
553    let p = headers.len();
554    let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
555    if open_ms > 100.0 {
556        log::info!(
557            "[DATA-LOAD] delim_open+headers | n_headers={} | n_proj={} | {:.1}ms",
558            all_headers.len(),
559            p,
560            open_ms
561        );
562    }
563
564    // Pass 1 discovers only the schema and row count. Keeping one inference
565    // state per column (instead of one owned String per cell) makes peak memory
566    // independent of the file's textual representation.
567    let mut inference = vec![DelimitedInferenceState::default(); p];
568    let mut total_rows: usize = 0;
569    let t_stream = std::time::Instant::now();
570    let mut record = StringRecord::new();
571    while rdr
572        .read_record(&mut record)
573        .map_err(|e| DataError::ParseError {
574            reason: format!("failed reading row: {e}"),
575        })?
576    {
577        if record.len() != all_headers.len() {
578            return Err(DataError::SchemaMismatch {
579                reason: format!(
580                    "row width mismatch at row {}: got {} fields, expected {}",
581                    total_rows + 1,
582                    record.len(),
583                    all_headers.len()
584                ),
585            });
586        }
587        total_rows += 1;
588        for (j, &selected_idx) in selected_indices.iter().enumerate() {
589            inference[j].observe(
590                record
591                    .get(selected_idx)
592                    .expect("record width was checked against the header row above")
593                    .trim(),
594                total_rows,
595                &headers[j],
596            )?;
597        }
598    }
599
600    let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
601    if stream_ms > 100.0 {
602        log::info!(
603            "[DATA-LOAD] delim_stream | n_rows={} | n_cols={} | {:.1}ms",
604            total_rows,
605            p,
606            stream_ms
607        );
608    }
609
610    if total_rows == 0 {
611        return Err(DataError::EmptyInput {
612            reason: "file has no rows".to_string(),
613        });
614    }
615
616    let t_schema = std::time::Instant::now();
617    let column_kinds = inference
618        .iter()
619        .enumerate()
620        .map(|(j, state)| state.kind(categorical_roles.contains(headers[j].as_str())))
621        .collect::<Vec<_>>();
622    let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
623    if schema_ms > 100.0 {
624        let n_cat = column_kinds
625            .iter()
626            .filter(|k| matches!(k, ColumnKindTag::Categorical))
627            .count();
628        log::info!(
629            "[DATA-LOAD] delim_convert+infer | n_cols={} | n_cat={} | {:.1}ms",
630            p,
631            n_cat,
632            schema_ms
633        );
634    }
635
636    // Pass 2 parses directly into the one final allocation. Categorical cells
637    // receive provisional encounter-order codes while only the UNIQUE level
638    // strings remain resident; a final in-place remap establishes canonical
639    // sorted codes without retaining a per-cell string table.
640    let t_assemble = std::time::Instant::now();
641    let mut values = Array2::<f64>::zeros((total_rows, p));
642    let mut categorical_encoders = (0..p)
643        .map(|j| {
644            matches!(column_kinds[j], ColumnKindTag::Categorical).then(CategoricalEncoder::default)
645        })
646        .collect::<Vec<_>>();
647    let mut encode_rdr = ReaderBuilder::new()
648        .has_headers(true)
649        .delimiter(delimiter)
650        .from_path(path)
651        .map_err(|e| DataError::ParseError {
652            reason: format!("failed to reopen '{}': {e}", path.display()),
653        })?;
654    encode_rdr.headers().map_err(|e| DataError::ParseError {
655        reason: format!("failed to reread headers: {e}"),
656    })?;
657    let mut encoded_rows = 0usize;
658    while encode_rdr
659        .read_record(&mut record)
660        .map_err(|e| DataError::ParseError {
661            reason: format!("failed reading row: {e}"),
662        })?
663    {
664        if record.len() != all_headers.len() {
665            return Err(DataError::SchemaMismatch {
666                reason: format!(
667                    "row width mismatch at row {}: got {} fields, expected {}",
668                    encoded_rows + 1,
669                    record.len(),
670                    all_headers.len()
671                ),
672            });
673        }
674        if encoded_rows >= total_rows {
675            return Err(DataError::SchemaMismatch {
676                reason: "data file changed while its schema was being discovered".to_string(),
677            });
678        }
679        for (j, &selected_idx) in selected_indices.iter().enumerate() {
680            let raw = record
681                .get(selected_idx)
682                .expect("record width was checked against the header row above")
683                .trim();
684            values[[encoded_rows, j]] = match column_kinds[j] {
685                ColumnKindTag::Continuous | ColumnKindTag::Binary => {
686                    parse_inferred_numeric_cell(raw, encoded_rows + 1, &headers[j])?
687                }
688                ColumnKindTag::Categorical => {
689                    if raw.is_empty() {
690                        return Err(DataError::EmptyInput {
691                            reason: format!(
692                                "empty field at row {}, column '{}'",
693                                encoded_rows + 1,
694                                &headers[j]
695                            ),
696                        });
697                    }
698                    categorical_encoders[j]
699                        .as_mut()
700                        .expect("categorical encoder")
701                        .encode(raw) as f64
702                }
703            };
704        }
705        encoded_rows += 1;
706    }
707    if encoded_rows != total_rows {
708        return Err(DataError::SchemaMismatch {
709            reason: "data file changed while its schema was being discovered".to_string(),
710        });
711    }
712
713    let mut levels = vec![Vec::<String>::new(); p];
714    for (j, encoder) in categorical_encoders.into_iter().enumerate() {
715        if let Some(encoder) = encoder {
716            levels[j] = encoder.finish(values.column_mut(j), LevelOrder::Canonical);
717        }
718    }
719    let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
720    if assemble_ms > 100.0 {
721        log::info!(
722            "[DATA-LOAD] delim_assemble_array2 | n_rows={} | n_cols={} | {:.1}ms",
723            total_rows,
724            p,
725            assemble_ms
726        );
727    }
728
729    let schema = DataSchema {
730        columns: headers
731            .iter()
732            .enumerate()
733            .map(|(j, name)| SchemaColumn {
734                name: name.clone(),
735                kind: column_kinds[j],
736                levels: std::mem::take(&mut levels[j]),
737            })
738            .collect(),
739    };
740    Ok(EncodedDataset {
741        headers,
742        values,
743        schema,
744        column_kinds,
745    })
746}
747
748#[derive(Clone, Copy)]
749struct DelimitedInferenceState {
750    all_numeric: bool,
751    all_binary: bool,
752    /// At least one cell was actually present and parsed as a number. Without
753    /// it an all-missing column would keep `all_numeric` vacuously true and
754    /// become an all-NaN Binary column (#2495).
755    saw_numeric: bool,
756}
757
758impl Default for DelimitedInferenceState {
759    fn default() -> Self {
760        Self {
761            all_numeric: true,
762            all_binary: true,
763            saw_numeric: false,
764        }
765    }
766}
767
768impl DelimitedInferenceState {
769    fn observe(&mut self, raw: &str, row: usize, header: &str) -> Result<(), DataError> {
770        if raw.is_empty() {
771            return Err(DataError::EmptyInput {
772                reason: format!("empty field at row {row}, column '{header}'"),
773            });
774        }
775        // #2495: a missing marker is neither a number nor a level — it must not
776        // drag the column out of `all_numeric` and into a factor over its own
777        // measurements. Same rule as `infer_schema_column`.
778        if is_missing_marker(raw) {
779            return Ok(());
780        }
781        match raw.parse::<f64>() {
782            Ok(value) => {
783                self.saw_numeric = true;
784                if !value.is_finite() {
785                    return Err(DataError::InvalidValue {
786                        reason: format!("non-finite value at row {row}, column '{header}'"),
787                    });
788                }
789                if (value - 0.0).abs() >= 1e-12 && (value - 1.0).abs() >= 1e-12 {
790                    self.all_binary = false;
791                }
792            }
793            Err(_) => {
794                self.all_numeric = false;
795                self.all_binary = false;
796            }
797        }
798        Ok(())
799    }
800
801    fn kind(self, force_categorical: bool) -> ColumnKindTag {
802        if force_categorical || !(self.all_numeric && self.saw_numeric) {
803            ColumnKindTag::Categorical
804        } else if self.all_binary {
805            ColumnKindTag::Binary
806        } else {
807            ColumnKindTag::Continuous
808        }
809    }
810}
811
812/// Is this cell a MISSING-VALUE marker rather than a value?
813///
814/// The tokens are the ones the tools gam ingests from actually write for a
815/// missing cell: `NA` is what R's `write.csv` emits and is by far the most
816/// common in practice; `N/A` comes out of spreadsheet exports; `NULL` out of SQL
817/// dumps. Matched case-insensitively because the same tools disagree on case.
818///
819/// Deliberately NOT here:
820/// * the empty field — it is already a hard `EmptyInput` error on every path,
821///   and relaxing that is a separate decision with its own blast radius;
822/// * `NaN` / `inf` — those *do* parse via `f64::from_str`, so they already hit
823///   the `!value.is_finite()` guard and raise `InvalidValue`. That is loud and
824///   correct, and must stay that way.
825///
826/// The silent-wrong-data class this exists to kill is exactly the tokens that
827/// neither parse as a number nor are a genuine category (#2495).
828fn is_missing_marker(raw: &str) -> bool {
829    matches!(
830        raw.trim().to_ascii_uppercase().as_str(),
831        "NA" | "N/A" | "NULL"
832    )
833}
834
835fn parse_inferred_numeric_cell(raw: &str, row: usize, header: &str) -> Result<f64, DataError> {
836    if raw.is_empty() {
837        return Err(DataError::EmptyInput {
838            reason: format!("empty field at row {row}, column '{header}'"),
839        });
840    }
841    // A missing cell in a numeric column IS the missing value, not a parse
842    // failure. NaN is the representation callers already assume when they filter
843    // with `is_finite()`; before #2495 the column was re-typed categorical and
844    // the level index was handed back as the measurement.
845    if is_missing_marker(raw) {
846        return Ok(f64::NAN);
847    }
848    let value = raw
849        .parse::<f64>()
850        .map_err(|error| DataError::EncodingFailure {
851            reason: format!(
852                "failed to parse numeric value '{raw}' at row {row}, column '{header}': {error}"
853            ),
854        })?;
855    if !value.is_finite() {
856        return Err(DataError::InvalidValue {
857            reason: format!("non-finite value at row {row}, column '{header}'"),
858        });
859    }
860    Ok(value)
861}
862
863#[derive(Clone, Copy)]
864enum LevelOrder {
865    Encounter,
866    Canonical,
867}
868
869/// Single-owner categorical encoder used by the file loaders.
870///
871/// The map owns each distinct label exactly once and assigns an encounter-order
872/// code while rows stream into the final numeric matrix. Finalization consumes
873/// the map, moving those same `String` allocations into the schema and, when a
874/// canonical order is required, remapping only the numeric codes in place.
875/// This avoids the former map + level-vector + canonicalization-clone triple
876/// ownership, whose text payload became row-sized for high-cardinality factors.
877#[derive(Default)]
878struct CategoricalEncoder {
879    encounter_codes: HashMap<String, usize>,
880}
881
882impl CategoricalEncoder {
883    fn encode(&mut self, label: &str) -> usize {
884        if let Some(&code) = self.encounter_codes.get(label) {
885            return code;
886        }
887        let code = self.encounter_codes.len();
888        self.encounter_codes.insert(label.to_owned(), code);
889        code
890    }
891
892    fn finish(self, mut encoded: ArrayViewMut1<'_, f64>, order: LevelOrder) -> Vec<String> {
893        match order {
894            LevelOrder::Encounter => {
895                let mut levels = std::iter::repeat_with(|| None)
896                    .take(self.encounter_codes.len())
897                    .collect::<Vec<Option<String>>>();
898                for (level, old_code) in self.encounter_codes {
899                    levels[old_code] = Some(level);
900                }
901                levels
902                    .into_iter()
903                    .map(|level| level.expect("encounter code must name one level"))
904                    .collect()
905            }
906            LevelOrder::Canonical => {
907                let mut levels_with_old_codes =
908                    self.encounter_codes.into_iter().collect::<Vec<_>>();
909                levels_with_old_codes
910                    .sort_by(|(a, _), (b, _)| natural_level_cmp(a.as_str(), b.as_str()));
911                let mut remap = vec![0usize; levels_with_old_codes.len()];
912                for (new_code, (_, old_code)) in levels_with_old_codes.iter().enumerate() {
913                    remap[*old_code] = new_code;
914                }
915                for code in encoded.iter_mut() {
916                    *code = remap[*code as usize] as f64;
917                }
918                levels_with_old_codes
919                    .into_iter()
920                    .map(|(level, _)| level)
921                    .collect()
922            }
923        }
924    }
925}
926
927fn load_delimited_with_schema(
928    path: &Path,
929    delimiter: u8,
930    schema: &DataSchema,
931    unseen_policy: UnseenCategoryPolicy,
932    requested_columns: &[String],
933) -> Result<EncodedDataset, DataError> {
934    let t_open = std::time::Instant::now();
935    let mut rdr = ReaderBuilder::new()
936        .has_headers(true)
937        .delimiter(delimiter)
938        .from_path(path)
939        .map_err(|e| DataError::ParseError {
940            reason: format!("failed to open '{}': {e}", path.display()),
941        })?;
942
943    let all_headers: Vec<String> = rdr
944        .headers()
945        .map_err(|e| DataError::ParseError {
946            reason: format!("failed to read headers: {e}"),
947        })?
948        .iter()
949        .map(|s| s.trim().to_string())
950        .collect();
951    if all_headers.is_empty() {
952        return Err(DataError::EmptyInput {
953            reason: "file has no headers".to_string(),
954        });
955    }
956    let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
957    let headers = projected_headers(&all_headers, &selected_indices);
958    let p = headers.len();
959    let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
960    if open_ms > 100.0 {
961        log::info!(
962            "[DATA-LOAD] delim_schema_open+headers | n_headers={} | n_proj={} | {:.1}ms",
963            all_headers.len(),
964            p,
965            open_ms
966        );
967    }
968
969    // Build per-column metadata from schema.
970    let schema_byname: HashMap<&str, &SchemaColumn> = schema
971        .columns
972        .iter()
973        .map(|c| (c.name.as_str(), c))
974        .collect();
975
976    let mut col_meta = Vec::<ColMeta>::with_capacity(p);
977    for name in &headers {
978        if let Some(sc) = schema_byname.get(name.as_str()) {
979            let level_map = if matches!(sc.kind, ColumnKindTag::Categorical) {
980                Some(
981                    sc.levels
982                        .iter()
983                        .enumerate()
984                        .map(|(idx, v)| (v.as_str(), idx as f64))
985                        .collect::<HashMap<_, _>>(),
986                )
987            } else {
988                None
989            };
990            col_meta.push(ColMeta {
991                kind: sc.kind,
992                level_map,
993                schema_col: (*sc).clone(),
994            });
995        } else {
996            // Column not in schema — will be inferred below (fallback).
997            col_meta.push(ColMeta {
998                kind: ColumnKindTag::Continuous, // tentative
999                level_map: None,
1000                schema_col: SchemaColumn {
1001                    name: name.clone(),
1002                    kind: ColumnKindTag::Continuous,
1003                    levels: Vec::new(),
1004                },
1005            });
1006        }
1007    }
1008
1009    // Track which columns need inference (not in provided schema).
1010    let needs_inference: Vec<bool> = headers
1011        .iter()
1012        .map(|h| !schema_byname.contains_key(h.as_str()))
1013        .collect();
1014
1015    // A complete supplied schema needs no discovery pass. Parse its rows once
1016    // in historical row-major error order into a growable row-major buffer;
1017    // `Array2::from_shape_vec` adopts that same allocation as the final matrix.
1018    // The two-pass path below remains only for projected columns absent from the
1019    // schema, whose kind genuinely has to be discovered before encoding.
1020    if needs_inference.iter().all(|needs| !needs) {
1021        let t_stream = std::time::Instant::now();
1022        let mut flat_values = Vec::<f64>::new();
1023        let mut total_rows = 0usize;
1024        let mut record = StringRecord::new();
1025        while rdr
1026            .read_record(&mut record)
1027            .map_err(|e| DataError::ParseError {
1028                reason: format!("failed reading row: {e}"),
1029            })?
1030        {
1031            if record.len() != all_headers.len() {
1032                return Err(DataError::SchemaMismatch {
1033                    reason: format!(
1034                        "row width mismatch at row {}: got {} fields, expected {}",
1035                        total_rows + 1,
1036                        record.len(),
1037                        all_headers.len()
1038                    ),
1039                });
1040            }
1041            total_rows += 1;
1042            for j in 0..p {
1043                let raw = record
1044                    .get(selected_indices[j])
1045                    .expect("record width was checked against the header row above")
1046                    .trim();
1047                flat_values.push(parse_cell_with_schema(
1048                    raw,
1049                    &col_meta[j],
1050                    total_rows,
1051                    &headers[j],
1052                    &unseen_policy,
1053                )?);
1054            }
1055        }
1056        if total_rows == 0 {
1057            return Err(DataError::EmptyInput {
1058                reason: "file has no rows".to_string(),
1059            });
1060        }
1061        let values = Array2::from_shape_vec((total_rows, p), flat_values).map_err(|error| {
1062            DataError::EncodingFailure {
1063                reason: format!("failed to assemble schema-guided delimited matrix: {error}"),
1064            }
1065        })?;
1066        let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
1067        if stream_ms > 100.0 {
1068            log::info!(
1069                "[DATA-LOAD] delim_schema_direct | n_rows={} | n_cols={} | {:.1}ms",
1070                total_rows,
1071                p,
1072                stream_ms
1073            );
1074        }
1075        let column_kinds = col_meta.iter().map(|meta| meta.kind).collect();
1076        let schema_out = DataSchema {
1077            columns: col_meta.into_iter().map(|meta| meta.schema_col).collect(),
1078        };
1079        return Ok(EncodedDataset {
1080            headers,
1081            values,
1082            schema: schema_out,
1083            column_kinds,
1084        });
1085    }
1086
1087    // Pass 1 validates schema-bound cells, discovers the kinds of schema-less
1088    // columns, and counts rows. No cell payload survives this pass.
1089    let mut inference = vec![DelimitedInferenceState::default(); p];
1090    let mut total_rows: usize = 0;
1091    let t_stream = std::time::Instant::now();
1092    let mut record = StringRecord::new();
1093    while rdr
1094        .read_record(&mut record)
1095        .map_err(|e| DataError::ParseError {
1096            reason: format!("failed reading row: {e}"),
1097        })?
1098    {
1099        if record.len() != all_headers.len() {
1100            return Err(DataError::SchemaMismatch {
1101                reason: format!(
1102                    "row width mismatch at row {}: got {} fields, expected {}",
1103                    total_rows + 1,
1104                    record.len(),
1105                    all_headers.len()
1106                ),
1107            });
1108        }
1109        total_rows += 1;
1110
1111        for j in 0..p {
1112            let raw = record
1113                    .get(selected_indices[j])
1114                    .expect("record width was checked against the header row above")
1115                    .trim();
1116            if needs_inference[j] {
1117                inference[j].observe(raw, total_rows, &headers[j])?;
1118            } else {
1119                parse_cell_with_schema(raw, &col_meta[j], total_rows, &headers[j], &unseen_policy)?;
1120            }
1121        }
1122    }
1123
1124    let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
1125    if stream_ms > 100.0 {
1126        let n_inf = needs_inference.iter().filter(|x| **x).count();
1127        log::info!(
1128            "[DATA-LOAD] delim_schema_stream | n_rows={} | n_cols={} | n_inf={} | {:.1}ms",
1129            total_rows,
1130            p,
1131            n_inf,
1132            stream_ms
1133        );
1134    }
1135
1136    if total_rows == 0 {
1137        return Err(DataError::EmptyInput {
1138            reason: "file has no rows".to_string(),
1139        });
1140    }
1141
1142    let t_finalize = std::time::Instant::now();
1143    for j in 0..p {
1144        if needs_inference[j] {
1145            let kind = inference[j].kind(false);
1146            col_meta[j].kind = kind;
1147            col_meta[j].schema_col.kind = kind;
1148        }
1149    }
1150    let finalize_ms = t_finalize.elapsed().as_secs_f64() * 1000.0;
1151    if finalize_ms > 100.0 {
1152        log::info!(
1153            "[DATA-LOAD] delim_schema_finalize | n_cols={} | {:.1}ms",
1154            p,
1155            finalize_ms
1156        );
1157    }
1158
1159    // Pass 2 parses directly into final storage. Only unique strings for newly
1160    // inferred categorical columns are retained until their in-place canonical
1161    // code remap.
1162    let t_assemble = std::time::Instant::now();
1163    let mut values = Array2::<f64>::zeros((total_rows, p));
1164    let mut inferred_encoders = (0..p)
1165        .map(|j| {
1166            (needs_inference[j] && matches!(col_meta[j].kind, ColumnKindTag::Categorical))
1167                .then(CategoricalEncoder::default)
1168        })
1169        .collect::<Vec<_>>();
1170    let mut encode_rdr = ReaderBuilder::new()
1171        .has_headers(true)
1172        .delimiter(delimiter)
1173        .from_path(path)
1174        .map_err(|e| DataError::ParseError {
1175            reason: format!("failed to reopen '{}': {e}", path.display()),
1176        })?;
1177    encode_rdr.headers().map_err(|e| DataError::ParseError {
1178        reason: format!("failed to reread headers: {e}"),
1179    })?;
1180    let mut encoded_rows = 0usize;
1181    while encode_rdr
1182        .read_record(&mut record)
1183        .map_err(|e| DataError::ParseError {
1184            reason: format!("failed reading row: {e}"),
1185        })?
1186    {
1187        if record.len() != all_headers.len() {
1188            return Err(DataError::SchemaMismatch {
1189                reason: format!(
1190                    "row width mismatch at row {}: got {} fields, expected {}",
1191                    encoded_rows + 1,
1192                    record.len(),
1193                    all_headers.len()
1194                ),
1195            });
1196        }
1197        if encoded_rows >= total_rows {
1198            return Err(DataError::SchemaMismatch {
1199                reason: "data file changed while its schema was being discovered".to_string(),
1200            });
1201        }
1202        for j in 0..p {
1203            let raw = record
1204                    .get(selected_indices[j])
1205                    .expect("record width was checked against the header row above")
1206                    .trim();
1207            values[[encoded_rows, j]] = if !needs_inference[j] {
1208                parse_cell_with_schema(
1209                    raw,
1210                    &col_meta[j],
1211                    encoded_rows + 1,
1212                    &headers[j],
1213                    &unseen_policy,
1214                )?
1215            } else {
1216                match col_meta[j].kind {
1217                    ColumnKindTag::Continuous | ColumnKindTag::Binary => {
1218                        parse_inferred_numeric_cell(raw, encoded_rows + 1, &headers[j])?
1219                    }
1220                    ColumnKindTag::Categorical => {
1221                        if raw.is_empty() {
1222                            return Err(DataError::EmptyInput {
1223                                reason: format!(
1224                                    "empty field at row {}, column '{}'",
1225                                    encoded_rows + 1,
1226                                    &headers[j]
1227                                ),
1228                            });
1229                        }
1230                        let encoder = inferred_encoders[j]
1231                            .as_mut()
1232                            .expect("inferred categorical encoder");
1233                        encoder.encode(raw) as f64
1234                    }
1235                }
1236            };
1237        }
1238        encoded_rows += 1;
1239    }
1240    if encoded_rows != total_rows {
1241        return Err(DataError::SchemaMismatch {
1242            reason: "data file changed while its schema was being discovered".to_string(),
1243        });
1244    }
1245    for (j, encoder) in inferred_encoders.into_iter().enumerate() {
1246        if let Some(encoder) = encoder {
1247            col_meta[j].schema_col.levels =
1248                encoder.finish(values.column_mut(j), LevelOrder::Canonical);
1249        }
1250    }
1251    let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
1252    if assemble_ms > 100.0 {
1253        log::info!(
1254            "[DATA-LOAD] delim_schema_assemble | n_rows={} | n_cols={} | {:.1}ms",
1255            total_rows,
1256            p,
1257            assemble_ms
1258        );
1259    }
1260
1261    let column_kinds = col_meta.iter().map(|meta| meta.kind).collect();
1262    let schema_out = DataSchema {
1263        columns: col_meta.into_iter().map(|m| m.schema_col).collect(),
1264    };
1265    Ok(EncodedDataset {
1266        headers,
1267        values,
1268        schema: schema_out,
1269        column_kinds,
1270    })
1271}
1272
1273fn parse_cell_with_schema(
1274    raw: &str,
1275    meta: &ColMeta<'_>,
1276    row: usize,
1277    col_name: &str,
1278    unseen_policy: &UnseenCategoryPolicy,
1279) -> Result<f64, DataError> {
1280    let val = match meta.kind {
1281        // A schema-declared numeric column still has to accept the missing
1282        // marker as the missing value (#2495), not fail to parse it.
1283        ColumnKindTag::Continuous if is_missing_marker(raw) => f64::NAN,
1284        ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
1285            DataError::SchemaMismatch {
1286                reason: format!(
1287                    "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
1288                    col_name, row, raw, err
1289                ),
1290            }
1291        })?,
1292        ColumnKindTag::Binary if is_missing_marker(raw) => f64::NAN,
1293        ColumnKindTag::Binary => {
1294            let v = raw
1295                .parse::<f64>()
1296                .map_err(|err| DataError::SchemaMismatch {
1297                    reason: format!(
1298                        "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
1299                        col_name, row, raw, err
1300                    ),
1301                })?;
1302            if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
1303                return Err(DataError::SchemaMismatch {
1304                    reason: format!(
1305                        "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
1306                        col_name, row, v
1307                    ),
1308                });
1309            }
1310            v
1311        }
1312        ColumnKindTag::Categorical => {
1313            let map = meta
1314                .level_map
1315                .as_ref()
1316                .ok_or_else(|| DataError::EncodingFailure {
1317                    reason: "internal categorical schema map missing".to_string(),
1318                })?;
1319            match map.get(raw) {
1320                Some(v) => *v,
1321                None => unseen_policy
1322                    .unseen_code_for(col_name, meta.schema_col.levels.len())
1323                    .ok_or_else(|| DataError::SchemaMismatch {
1324                        reason: format!(
1325                            "unseen level '{}' in categorical column '{}' at row {}",
1326                            raw, col_name, row
1327                        ),
1328                    })?,
1329            }
1330        }
1331    };
1332    // The NaN produced for a missing marker (#2495) IS the encoded value here,
1333    // so it must not be re-rejected by the non-finite guard. Every other route
1334    // to a non-finite value still fails loudly.
1335    if !val.is_finite() && !is_missing_marker(raw) {
1336        return Err(DataError::InvalidValue {
1337            reason: format!("non-finite value at row {}, column '{}'", row, col_name),
1338        });
1339    }
1340    Ok(val)
1341}
1342
1343// Inner type used by load_delimited_with_schema; defined here to keep
1344// parse_cell_with_schema usable without forward-declaring inside the fn.
1345struct ColMeta<'a> {
1346    kind: ColumnKindTag,
1347    level_map: Option<HashMap<&'a str, f64>>,
1348    schema_col: SchemaColumn,
1349}
1350
1351// ---------------------------------------------------------------------------
1352// Parquet — columnar, zero StringRecord, schema from metadata
1353// ---------------------------------------------------------------------------
1354
1355/// True iff an Arrow column should be treated as a string/categorical column.
1356///
1357/// Dictionary encoding is a *storage* detail, not a semantic type: pyarrow
1358/// dictionary-encodes low-cardinality columns by default, including numeric
1359/// ones (integer factor levels, small enums stored as ints). A
1360/// `Dictionary(K, V)` column is categorical iff its *value* type `V` is a
1361/// string type; `Dictionary(_, Int*/UInt*/Float*/Bool)` is numeric. We recurse
1362/// through the value type so nested dictionaries resolve to their leaf type.
1363fn arrow_field_is_string(dt: &arrow::datatypes::DataType) -> bool {
1364    use arrow::datatypes::DataType;
1365    match dt {
1366        DataType::Utf8 | DataType::LargeUtf8 => true,
1367        DataType::Dictionary(_, value_type) => arrow_field_is_string(value_type),
1368        _ => false,
1369    }
1370}
1371
1372fn write_arrow_numeric_values(
1373    values: impl IntoIterator<Item = f64>,
1374    base_row: usize,
1375    header: &str,
1376    mut output: ArrayViewMut1<'_, f64>,
1377    categorical_encoder: Option<&mut CategoricalEncoder>,
1378    all_binary: &mut bool,
1379) -> Result<(), DataError> {
1380    match categorical_encoder {
1381        Some(encoder) => {
1382            for (batch_row, value) in values.into_iter().enumerate() {
1383                if !value.is_finite() {
1384                    return Err(DataError::InvalidValue {
1385                        reason: format!(
1386                            "non-finite value at row {}, column '{}'",
1387                            base_row + batch_row + 1,
1388                            header
1389                        ),
1390                    });
1391                }
1392                output[batch_row] = encoder.encode(&value.to_string()) as f64;
1393            }
1394        }
1395        None => {
1396            for (batch_row, value) in values.into_iter().enumerate() {
1397                if !value.is_finite() {
1398                    return Err(DataError::InvalidValue {
1399                        reason: format!(
1400                            "non-finite value at row {}, column '{}'",
1401                            base_row + batch_row + 1,
1402                            header
1403                        ),
1404                    });
1405                }
1406                if (value - 0.0).abs() >= 1e-12 && (value - 1.0).abs() >= 1e-12 {
1407                    *all_binary = false;
1408                }
1409                output[batch_row] = value;
1410            }
1411        }
1412    }
1413    Ok(())
1414}
1415
1416fn reject_arrow_null_values(
1417    col: &dyn arrow::array::Array,
1418    base_row: usize,
1419    header: &str,
1420) -> Result<(), DataError> {
1421    let Some(nulls) = col.logical_nulls() else {
1422        return Ok(());
1423    };
1424    if let Some(batch_row) = (0..col.len()).find(|&row| nulls.is_null(row)) {
1425        return Err(DataError::InvalidValue {
1426            reason: format!(
1427                "null value at row {}, column '{}'",
1428                base_row + batch_row + 1,
1429                header
1430            ),
1431        });
1432    }
1433    Ok(())
1434}
1435
1436fn arrow_dictionary_string_value_at<'a, K>(
1437    col: &'a dyn arrow::array::Array,
1438    index: usize,
1439    logical_row: usize,
1440    header: &str,
1441) -> Result<&'a str, DataError>
1442where
1443    K: arrow::datatypes::ArrowDictionaryKeyType,
1444{
1445    use arrow::array::DictionaryArray;
1446
1447    let dictionary = col
1448        .as_any()
1449        .downcast_ref::<DictionaryArray<K>>()
1450        .ok_or_else(|| DataError::EncodingFailure {
1451            reason: format!(
1452                "Arrow dictionary column '{}' did not match its declared key type",
1453                header
1454            ),
1455        })?;
1456    let value_index = dictionary
1457        .key(index)
1458        .ok_or_else(|| DataError::InvalidValue {
1459            reason: format!("null value at row {logical_row}, column '{header}'"),
1460        })?;
1461    if value_index >= dictionary.values().len() {
1462        return Err(DataError::EncodingFailure {
1463            reason: format!(
1464                "Arrow dictionary column '{}' has out-of-range key {} at row {}",
1465                header, value_index, logical_row
1466            ),
1467        });
1468    }
1469    arrow_string_value_at(
1470        dictionary.values().as_ref(),
1471        value_index,
1472        logical_row,
1473        header,
1474    )
1475}
1476
1477/// Resolve a string or (possibly nested) dictionary-string value without
1478/// materializing a row-sized decoded string array.
1479fn arrow_string_value_at<'a>(
1480    col: &'a dyn arrow::array::Array,
1481    index: usize,
1482    logical_row: usize,
1483    header: &str,
1484) -> Result<&'a str, DataError> {
1485    use arrow::array::{LargeStringArray, StringArray};
1486    use arrow::datatypes::{
1487        DataType, Int8Type, Int16Type, Int32Type, Int64Type, UInt8Type, UInt16Type, UInt32Type,
1488        UInt64Type,
1489    };
1490
1491    if index >= col.len() {
1492        return Err(DataError::EncodingFailure {
1493            reason: format!(
1494                "Arrow string column '{}' has out-of-range index {} at row {}",
1495                header, index, logical_row
1496            ),
1497        });
1498    }
1499    if col.is_null(index) {
1500        return Err(DataError::InvalidValue {
1501            reason: format!("null value at row {logical_row}, column '{header}'"),
1502        });
1503    }
1504
1505    match col.data_type() {
1506        DataType::Utf8 => col
1507            .as_any()
1508            .downcast_ref::<StringArray>()
1509            .map(|array| array.value(index))
1510            .ok_or_else(|| DataError::EncodingFailure {
1511                reason: format!("Arrow column '{}' could not be read as Utf8", header),
1512            }),
1513        DataType::LargeUtf8 => col
1514            .as_any()
1515            .downcast_ref::<LargeStringArray>()
1516            .map(|array| array.value(index))
1517            .ok_or_else(|| DataError::EncodingFailure {
1518                reason: format!("Arrow column '{}' could not be read as LargeUtf8", header),
1519            }),
1520        DataType::Dictionary(key_type, _) => match key_type.as_ref() {
1521            DataType::Int8 => {
1522                arrow_dictionary_string_value_at::<Int8Type>(col, index, logical_row, header)
1523            }
1524            DataType::Int16 => {
1525                arrow_dictionary_string_value_at::<Int16Type>(col, index, logical_row, header)
1526            }
1527            DataType::Int32 => {
1528                arrow_dictionary_string_value_at::<Int32Type>(col, index, logical_row, header)
1529            }
1530            DataType::Int64 => {
1531                arrow_dictionary_string_value_at::<Int64Type>(col, index, logical_row, header)
1532            }
1533            DataType::UInt8 => {
1534                arrow_dictionary_string_value_at::<UInt8Type>(col, index, logical_row, header)
1535            }
1536            DataType::UInt16 => {
1537                arrow_dictionary_string_value_at::<UInt16Type>(col, index, logical_row, header)
1538            }
1539            DataType::UInt32 => {
1540                arrow_dictionary_string_value_at::<UInt32Type>(col, index, logical_row, header)
1541            }
1542            DataType::UInt64 => {
1543                arrow_dictionary_string_value_at::<UInt64Type>(col, index, logical_row, header)
1544            }
1545            other => Err(DataError::InvalidValue {
1546                reason: format!(
1547                    "unsupported Arrow dictionary key type {:?} for column '{}'",
1548                    other, header
1549                ),
1550            }),
1551        },
1552        other => Err(DataError::InvalidValue {
1553            reason: format!(
1554                "unsupported Arrow string column type {:?} for column '{}'",
1555                other, header
1556            ),
1557        }),
1558    }
1559}
1560
1561/// Decode one Arrow column directly into its final matrix column.
1562///
1563/// The Arrow record batch remains the bounded input buffer. No Rust
1564/// `Vec<f64>` or per-cell `Vec<String>` is materialized beside it: primitive
1565/// values are converted into the strided ndarray view, and categorical labels
1566/// are interned immediately into their single-owner encoder.
1567fn decode_arrow_batch_column_into(
1568    col: &dyn arrow::array::Array,
1569    base_row: usize,
1570    header: &str,
1571    is_string_col: bool,
1572    mut output: ArrayViewMut1<'_, f64>,
1573    mut categorical_encoder: Option<&mut CategoricalEncoder>,
1574    all_binary: &mut bool,
1575) -> Result<(), DataError> {
1576    use arrow::array::{
1577        BooleanArray, Float32Array, Float64Array, Int8Array, Int16Array, Int32Array, Int64Array,
1578        UInt8Array, UInt16Array, UInt32Array, UInt64Array,
1579    };
1580    use arrow::datatypes::DataType;
1581
1582    let n_rows = output.len();
1583    if col.len() != n_rows {
1584        return Err(DataError::SchemaMismatch {
1585            reason: format!(
1586                "Arrow column '{}' has {} rows, but its record batch has {}",
1587                header,
1588                col.len(),
1589                n_rows
1590            ),
1591        });
1592    }
1593    reject_arrow_null_values(col, base_row, header)?;
1594
1595    if is_string_col {
1596        let encoder =
1597            categorical_encoder
1598                .as_deref_mut()
1599                .ok_or_else(|| DataError::EncodingFailure {
1600                    reason: format!("categorical Arrow encoder missing for column '{header}'"),
1601                })?;
1602        for batch_row in 0..n_rows {
1603            let label = arrow_string_value_at(col, batch_row, base_row + batch_row + 1, header)?;
1604            output[batch_row] = encoder.encode(label) as f64;
1605        }
1606        return Ok(());
1607    }
1608
1609    // Numeric-valued dictionary columns (pyarrow dictionary-encodes
1610    // low-cardinality numeric columns by default) are not directly a
1611    // primitive array. Decode them to their concrete value type so the normal
1612    // numeric arms below apply. `arrow_field_is_string` has already routed
1613    // string-valued dictionaries through the categorical branch above, so any
1614    // dictionary reaching here has a numeric value type.
1615    let decoded_col;
1616    let col: &dyn arrow::array::Array = if let DataType::Dictionary(_, value_type) = col.data_type()
1617    {
1618        decoded_col = arrow::compute::cast(col, value_type).map_err(|e| DataError::ParseError {
1619            reason: format!(
1620                "failed to decode dictionary-encoded numeric column '{}': {e}",
1621                header
1622            ),
1623        })?;
1624        decoded_col.as_ref()
1625    } else {
1626        col
1627    };
1628    // Defensively validate the decoded primitive too: the cast itself must not
1629    // introduce a null while resolving dictionary keys.
1630    reject_arrow_null_values(col, base_row, header)?;
1631
1632    macro_rules! write_primitive {
1633        ($array_type:ty, $convert:expr) => {{
1634            let array = col
1635                .as_any()
1636                .downcast_ref::<$array_type>()
1637                .expect("array type is the one this `col.data_type()` arm matched");
1638            write_arrow_numeric_values(
1639                array.values().iter().copied().map($convert),
1640                base_row,
1641                header,
1642                output,
1643                categorical_encoder,
1644                all_binary,
1645            )
1646        }};
1647    }
1648
1649    match col.data_type() {
1650        DataType::Float64 => write_primitive!(Float64Array, |value: f64| value),
1651        DataType::Float32 => write_primitive!(Float32Array, |value: f32| value as f64),
1652        DataType::Int64 => write_primitive!(Int64Array, |value: i64| value as f64),
1653        DataType::Int32 => write_primitive!(Int32Array, |value: i32| value as f64),
1654        DataType::Int16 => write_primitive!(Int16Array, |value: i16| value as f64),
1655        DataType::Int8 => write_primitive!(Int8Array, |value: i8| value as f64),
1656        DataType::UInt64 => write_primitive!(UInt64Array, |value: u64| value as f64),
1657        DataType::UInt32 => write_primitive!(UInt32Array, |value: u32| value as f64),
1658        DataType::UInt16 => write_primitive!(UInt16Array, |value: u16| value as f64),
1659        DataType::UInt8 => write_primitive!(UInt8Array, |value: u8| value as f64),
1660        DataType::Boolean => {
1661            let arr = col
1662                .as_any()
1663                .downcast_ref::<BooleanArray>()
1664                .expect("array type is BooleanArray in the DataType::Boolean arm");
1665            write_arrow_numeric_values(
1666                (0..n_rows).map(|i| if arr.value(i) { 1.0 } else { 0.0 }),
1667                base_row,
1668                header,
1669                output,
1670                categorical_encoder,
1671                all_binary,
1672            )
1673        }
1674        other => Err(DataError::InvalidValue {
1675            reason: format!(
1676                "unsupported Arrow column type {:?} for column '{}'",
1677                other, header
1678            ),
1679        }),
1680    }
1681}
1682
1683/// Infer and encode an Arrow record-batch stream into gam's native dataset.
1684///
1685/// `headers` supplies the already-normalized public column names in record-
1686/// batch column order. The Arrow schema supplies only physical types: integer,
1687/// unsigned integer, floating-point, and boolean columns remain numeric, while
1688/// `Utf8`, `LargeUtf8`, and string-valued dictionaries become categorical.
1689/// Categorical labels are interned as batches stream and canonicalized with the
1690/// same natural ordering as the other inferred ingestion paths.
1691///
1692/// The final row-major `N x P` `f64` allocation is grown in place and then
1693/// moved into [`EncodedDataset::values`] without a second dense allocation.
1694/// Apart from Arrow's current input batch, only one owned string per distinct
1695/// categorical level is retained; no row-major string table is materialized.
1696pub fn encode_arrow_record_batch_reader_with_inferred_schema(
1697    reader: &mut dyn arrow::record_batch::RecordBatchReader,
1698    headers: Vec<String>,
1699) -> Result<EncodedDataset, DataError> {
1700    if headers.is_empty() {
1701        return Err(DataError::EmptyInput {
1702            reason: "Arrow table must have at least one header column".to_string(),
1703        });
1704    }
1705
1706    let mut seen_headers = HashSet::<&str>::with_capacity(headers.len());
1707    for (column, header) in headers.iter().enumerate() {
1708        if header.trim().is_empty() {
1709            return Err(DataError::EmptyInput {
1710                reason: format!("Arrow header at column {} cannot be empty", column + 1),
1711            });
1712        }
1713        if !seen_headers.insert(header.as_str()) {
1714            return Err(DataError::SchemaMismatch {
1715                reason: format!("duplicate Arrow header '{}'", header),
1716            });
1717        }
1718    }
1719
1720    let arrow_schema = reader.schema();
1721    let p = headers.len();
1722    if arrow_schema.fields().len() != p {
1723        return Err(DataError::SchemaMismatch {
1724            reason: format!(
1725                "Arrow schema has {} columns, but {} normalized headers were supplied",
1726                arrow_schema.fields().len(),
1727                p
1728            ),
1729        });
1730    }
1731
1732    let is_string_col = arrow_schema
1733        .fields()
1734        .iter()
1735        .map(|field| arrow_field_is_string(field.data_type()))
1736        .collect::<Vec<_>>();
1737    let mut all_binary = vec![true; p];
1738    let mut categorical_encoders = is_string_col
1739        .iter()
1740        .map(|&is_string| is_string.then(CategoricalEncoder::default))
1741        .collect::<Vec<_>>();
1742    let mut encoded_values = Vec::<f64>::new();
1743    let mut rows_seen = 0usize;
1744
1745    for batch_result in reader {
1746        let batch = batch_result.map_err(|error| DataError::ParseError {
1747            reason: format!("failed to read Arrow record batch: {error}"),
1748        })?;
1749        if batch.num_columns() != p {
1750            return Err(DataError::SchemaMismatch {
1751                reason: format!(
1752                    "Arrow record batch has {} columns, but {} normalized headers were supplied",
1753                    batch.num_columns(),
1754                    p
1755                ),
1756            });
1757        }
1758        for j in 0..p {
1759            let expected = arrow_schema.field(j).data_type();
1760            let actual = batch.column(j).data_type();
1761            if actual != expected {
1762                return Err(DataError::SchemaMismatch {
1763                    reason: format!(
1764                        "Arrow column '{}' changed type between schema and batch: expected {:?}, got {:?}",
1765                        headers[j], expected, actual
1766                    ),
1767                });
1768            }
1769        }
1770
1771        let n_rows = batch.num_rows();
1772        let batch_values = n_rows
1773            .checked_mul(p)
1774            .ok_or_else(|| DataError::EncodingFailure {
1775                reason: "Arrow batch dimensions do not fit in memory address space".to_string(),
1776            })?;
1777        let next_len = encoded_values
1778            .len()
1779            .checked_add(batch_values)
1780            .ok_or_else(|| DataError::EncodingFailure {
1781                reason: "Arrow dataset dimensions do not fit in memory address space".to_string(),
1782            })?;
1783        encoded_values
1784            .try_reserve(batch_values)
1785            .map_err(|error| DataError::EncodingFailure {
1786                reason: format!("failed to reserve Arrow dataset storage: {error}"),
1787            })?;
1788        let batch_offset = encoded_values.len();
1789        encoded_values.resize(next_len, 0.0);
1790        let mut batch_output = ndarray::ArrayViewMut2::from_shape(
1791            (n_rows, p),
1792            &mut encoded_values[batch_offset..next_len],
1793        )
1794        .map_err(|error| DataError::EncodingFailure {
1795            reason: format!("failed to shape Arrow batch output: {error}"),
1796        })?;
1797
1798        let decoded_columns = batch_output
1799            .axis_iter_mut(Axis(1))
1800            .into_par_iter()
1801            .zip(categorical_encoders.par_iter_mut())
1802            .zip(all_binary.par_iter_mut())
1803            .enumerate()
1804            .map(|(j, ((output, encoder), column_all_binary))| {
1805                decode_arrow_batch_column_into(
1806                    batch.column(j).as_ref(),
1807                    rows_seen,
1808                    &headers[j],
1809                    is_string_col[j],
1810                    output,
1811                    encoder.as_mut(),
1812                    column_all_binary,
1813                )
1814            })
1815            .collect::<Vec<_>>();
1816        for decoded in decoded_columns {
1817            decoded?;
1818        }
1819        rows_seen = rows_seen
1820            .checked_add(n_rows)
1821            .ok_or_else(|| DataError::EncodingFailure {
1822                reason: "Arrow row count does not fit in memory address space".to_string(),
1823            })?;
1824    }
1825
1826    if rows_seen == 0 {
1827        return Err(DataError::EmptyInput {
1828            reason: "Arrow table data cannot be empty".to_string(),
1829        });
1830    }
1831
1832    let mut values = Array2::from_shape_vec((rows_seen, p), encoded_values).map_err(|error| {
1833        DataError::EncodingFailure {
1834            reason: format!("failed to shape encoded Arrow dataset: {error}"),
1835        }
1836    })?;
1837    let mut levels = vec![Vec::<String>::new(); p];
1838    for (j, encoder) in categorical_encoders.into_iter().enumerate() {
1839        if let Some(encoder) = encoder {
1840            levels[j] = encoder.finish(values.column_mut(j), LevelOrder::Canonical);
1841        }
1842    }
1843
1844    let mut schema_columns = Vec::<SchemaColumn>::with_capacity(p);
1845    let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
1846    for (j, name) in headers.iter().enumerate() {
1847        let kind = if is_string_col[j] {
1848            ColumnKindTag::Categorical
1849        } else if all_binary[j] {
1850            ColumnKindTag::Binary
1851        } else {
1852            ColumnKindTag::Continuous
1853        };
1854        column_kinds.push(kind);
1855        schema_columns.push(SchemaColumn {
1856            name: name.clone(),
1857            kind,
1858            levels: std::mem::take(&mut levels[j]),
1859        });
1860    }
1861
1862    Ok(EncodedDataset {
1863        headers,
1864        values,
1865        schema: DataSchema {
1866            columns: schema_columns,
1867        },
1868        column_kinds,
1869    })
1870}
1871
1872fn load_parquet_inferred(
1873    path: &Path,
1874    requested_columns: &[String],
1875    categorical_roles: &HashSet<&str>,
1876) -> Result<EncodedDataset, DataError> {
1877    use parquet::arrow::{ProjectionMask, arrow_reader::ParquetRecordBatchReaderBuilder};
1878    use rayon::prelude::*;
1879    use std::fs::File;
1880
1881    let t_open = std::time::Instant::now();
1882    let file = File::open(path).map_err(|e| DataError::ParseError {
1883        reason: format!("failed to open parquet '{}': {e}", path.display()),
1884    })?;
1885    let builder =
1886        ParquetRecordBatchReaderBuilder::try_new(file).map_err(|e| DataError::ParseError {
1887            reason: format!("failed to read parquet metadata '{}': {e}", path.display()),
1888        })?;
1889
1890    let full_schema = builder.schema().clone();
1891    let all_headers: Vec<String> = full_schema
1892        .fields()
1893        .iter()
1894        .map(|f| f.name().clone())
1895        .collect();
1896    if all_headers.is_empty() {
1897        return Err(DataError::EmptyInput {
1898            reason: "parquet file has no columns".to_string(),
1899        });
1900    }
1901    let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
1902    let headers = projected_headers(&all_headers, &selected_indices);
1903    let selected_fields = selected_indices
1904        .iter()
1905        .map(|&idx| full_schema.fields()[idx].clone())
1906        .collect::<Vec<_>>();
1907    let total_rows =
1908        usize::try_from(builder.metadata().file_metadata().num_rows()).map_err(|_| {
1909            DataError::ParseError {
1910                reason: "parquet row count does not fit in memory address space".to_string(),
1911            }
1912        })?;
1913    if total_rows == 0 {
1914        return Err(DataError::EmptyInput {
1915            reason: "parquet file has no rows".to_string(),
1916        });
1917    }
1918    let projection =
1919        ProjectionMask::roots(builder.parquet_schema(), selected_indices.iter().copied());
1920    let reader =
1921        builder
1922            .with_projection(projection)
1923            .build()
1924            .map_err(|e| DataError::ParseError {
1925                reason: format!("failed to build parquet reader: {e}"),
1926            })?;
1927    let p = headers.len();
1928    let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
1929    if open_ms > 100.0 {
1930        log::info!(
1931            "[DATA-LOAD] parquet_open+meta | n_headers={} | n_proj={} | {:.1}ms",
1932            all_headers.len(),
1933            p,
1934            open_ms
1935        );
1936    }
1937
1938    let t_batches = std::time::Instant::now();
1939    let is_string_col = selected_fields
1940        .iter()
1941        .map(|field| arrow_field_is_string(field.data_type()))
1942        .collect::<Vec<_>>();
1943    let forced_numeric_categorical = headers
1944        .iter()
1945        .enumerate()
1946        .map(|(j, header)| !is_string_col[j] && categorical_roles.contains(header.as_str()))
1947        .collect::<Vec<_>>();
1948    let mut values = Array2::<f64>::zeros((total_rows, p));
1949    let mut all_binary = vec![true; p];
1950    let mut categorical_encoders = (0..p)
1951        .map(|j| {
1952            (is_string_col[j] || forced_numeric_categorical[j]).then(CategoricalEncoder::default)
1953        })
1954        .collect::<Vec<_>>();
1955    let mut rows_seen = 0usize;
1956    for batch_result in reader {
1957        let batch = batch_result.map_err(|e| DataError::ParseError {
1958            reason: format!("failed to read parquet record batch: {e}"),
1959        })?;
1960        let n_rows = batch.num_rows();
1961        if rows_seen.saturating_add(n_rows) > total_rows {
1962            return Err(DataError::SchemaMismatch {
1963                reason: "parquet row count changed while reading record batches".to_string(),
1964            });
1965        }
1966
1967        let decoded_columns = values
1968            .slice_mut(s![rows_seen..rows_seen + n_rows, ..])
1969            .axis_iter_mut(Axis(1))
1970            .into_par_iter()
1971            .zip(categorical_encoders.par_iter_mut())
1972            .zip(all_binary.par_iter_mut())
1973            .enumerate()
1974            .map(|(j, ((output, encoder), column_all_binary))| {
1975                decode_arrow_batch_column_into(
1976                    batch.column(j).as_ref(),
1977                    rows_seen,
1978                    &headers[j],
1979                    is_string_col[j],
1980                    output,
1981                    encoder.as_mut(),
1982                    column_all_binary,
1983                )
1984            })
1985            .collect::<Vec<_>>();
1986
1987        // Rayon preserves indexed collection order, so checking the compact
1988        // per-column result vector from left to right retains the loader's
1989        // historical lowest-column error precedence without retaining decoded
1990        // cell payloads.
1991        for decoded in decoded_columns {
1992            decoded?;
1993        }
1994        rows_seen += n_rows;
1995    }
1996
1997    if rows_seen != total_rows {
1998        return Err(DataError::SchemaMismatch {
1999            reason: format!(
2000                "parquet metadata reports {total_rows} rows but record batches yielded {rows_seen}"
2001            ),
2002        });
2003    }
2004    let batches_ms = t_batches.elapsed().as_secs_f64() * 1000.0;
2005    if batches_ms > 100.0 {
2006        log::info!(
2007            "[DATA-LOAD] parquet_batches_decode | n_rows={} | n_cols={} | {:.1}ms",
2008            total_rows,
2009            p,
2010            batches_ms
2011        );
2012    }
2013    let t_schema = std::time::Instant::now();
2014    // Numeric factor roles use canonical sorted label order, matching the CSV
2015    // factor-by-construction path. String parquet columns retain their existing
2016    // encounter-order contract. Finalization consumes each label map, moving its
2017    // sole String allocation into the schema and remapping codes in place only
2018    // for the canonical case.
2019    let mut levels = vec![Vec::<String>::new(); p];
2020    for (j, encoder) in categorical_encoders.into_iter().enumerate() {
2021        if let Some(encoder) = encoder {
2022            let order = if forced_numeric_categorical[j] {
2023                LevelOrder::Canonical
2024            } else {
2025                LevelOrder::Encounter
2026            };
2027            levels[j] = encoder.finish(values.column_mut(j), order);
2028        }
2029    }
2030    let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
2031    let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2032    for j in 0..p {
2033        let kind = if is_string_col[j] || forced_numeric_categorical[j] {
2034            ColumnKindTag::Categorical
2035        } else if all_binary[j] {
2036            ColumnKindTag::Binary
2037        } else {
2038            ColumnKindTag::Continuous
2039        };
2040        column_kinds.push(kind);
2041        schema_cols.push(SchemaColumn {
2042            name: headers[j].clone(),
2043            kind,
2044            levels: std::mem::take(&mut levels[j]),
2045        });
2046    }
2047    let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
2048    if schema_ms > 100.0 {
2049        let n_cat = column_kinds
2050            .iter()
2051            .filter(|k| matches!(k, ColumnKindTag::Categorical))
2052            .count();
2053        log::info!(
2054            "[DATA-LOAD] parquet_finalize_schema | n_cols={} | n_cat={} | {:.1}ms",
2055            p,
2056            n_cat,
2057            schema_ms
2058        );
2059    }
2060
2061    Ok(EncodedDataset {
2062        headers,
2063        values,
2064        schema: DataSchema {
2065            columns: schema_cols,
2066        },
2067        column_kinds,
2068    })
2069}
2070
2071fn load_parquet_with_schema(
2072    path: &Path,
2073    schema: &DataSchema,
2074    unseen_policy: UnseenCategoryPolicy,
2075    requested_columns: &[String],
2076) -> Result<EncodedDataset, DataError> {
2077    // Load with inference first, then validate/re-encode against provided schema.
2078    // No formula roles are threaded here: the saved schema already records each
2079    // column's categorical kind, and the re-encode pass below pins kinds to it.
2080    let inferred = load_parquet_inferred(path, requested_columns, &HashSet::new())?;
2081    let p = inferred.headers.len();
2082    let n = inferred.values.nrows();
2083
2084    let schema_byname: HashMap<&str, &SchemaColumn> = schema
2085        .columns
2086        .iter()
2087        .map(|c| (c.name.as_str(), c))
2088        .collect();
2089
2090    let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2091    let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
2092    let mut values = inferred.values;
2093
2094    for j in 0..p {
2095        let name = &inferred.headers[j];
2096        if let Some(sc) = schema_byname.get(name.as_str()) {
2097            column_kinds.push(sc.kind);
2098            schema_cols.push((*sc).clone());
2099
2100            match sc.kind {
2101                ColumnKindTag::Continuous => {
2102                    if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2103                        return Err(DataError::SchemaMismatch {
2104                            reason: format!(
2105                                "column '{}' is continuous in schema but parquet column is string/categorical",
2106                                name
2107                            ),
2108                        });
2109                    }
2110                }
2111                ColumnKindTag::Binary => {
2112                    if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2113                        return Err(DataError::SchemaMismatch {
2114                            reason: format!(
2115                                "column '{}' is binary in schema but parquet column is string/categorical",
2116                                name
2117                            ),
2118                        });
2119                    }
2120                    // NaN marks a missing cell (#2495), not a 0/1 violation.
2121                    if let Some(row) = values.column(j).iter().position(|value| {
2122                        value.is_finite()
2123                            && (*value - 0.0).abs() >= 1e-12
2124                            && (*value - 1.0).abs() >= 1e-12
2125                    }) {
2126                        return Err(DataError::SchemaMismatch {
2127                            reason: format!(
2128                                "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2129                                name,
2130                                row + 1,
2131                                values[[row, j]]
2132                            ),
2133                        });
2134                    }
2135                }
2136                ColumnKindTag::Categorical => {
2137                    if !matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2138                        return Err(DataError::SchemaMismatch {
2139                            reason: format!(
2140                                "column '{}' is categorical in schema but parquet column is numeric",
2141                                name
2142                            ),
2143                        });
2144                    }
2145                    let inferred_col = &inferred.schema.columns[j];
2146                    // Build mapping: inferred_level_name -> schema_level_index.
2147                    let schema_level_map: HashMap<&str, f64> = sc
2148                        .levels
2149                        .iter()
2150                        .enumerate()
2151                        .map(|(idx, v)| (v.as_str(), idx as f64))
2152                        .collect();
2153                    let inferred_to_schema: Vec<f64> = inferred_col
2154                        .levels
2155                        .iter()
2156                        .map(|lv| {
2157                            schema_level_map
2158                                .get(lv.as_str())
2159                                .copied()
2160                                .or_else(|| unseen_policy.unseen_code_for(name, sc.levels.len()))
2161                                .ok_or_else(|| DataError::SchemaMismatch {
2162                                    reason: format!(
2163                                        "unseen level '{}' in categorical column '{}'",
2164                                        lv, name
2165                                    ),
2166                                })
2167                        })
2168                        .collect::<Result<Vec<_>, _>>()?;
2169                    for i in 0..n {
2170                        let old_code = values[[i, j]] as usize;
2171                        if old_code >= inferred_to_schema.len() {
2172                            let Some(unseen_code) =
2173                                unseen_policy.unseen_code_for(name, sc.levels.len())
2174                            else {
2175                                return Err(DataError::SchemaMismatch {
2176                                    reason: format!(
2177                                        "unseen categorical code at row {}, column '{}'",
2178                                        i + 1,
2179                                        name
2180                                    ),
2181                                });
2182                            };
2183                            values[[i, j]] = unseen_code;
2184                            continue;
2185                        }
2186                        values[[i, j]] = inferred_to_schema[old_code];
2187                    }
2188                }
2189            }
2190        } else {
2191            // Column not in schema — keep inferred.
2192            column_kinds.push(inferred.column_kinds[j]);
2193            schema_cols.push(inferred.schema.columns[j].clone());
2194        }
2195    }
2196
2197    Ok(EncodedDataset {
2198        headers: inferred.headers,
2199        values,
2200        schema: DataSchema {
2201            columns: schema_cols,
2202        },
2203        column_kinds,
2204    })
2205}
2206
2207pub fn encode_recordswith_inferred_schema(
2208    headers: Vec<String>,
2209    records: Vec<StringRecord>,
2210) -> Result<EncodedDataset, String> {
2211    if records.is_empty() {
2212        return Err(DataError::EmptyInput {
2213            reason: "table data cannot be empty".to_string(),
2214        }
2215        .into());
2216    }
2217    // Schema inference is column-independent: each column scans only its own
2218    // field across all rows. With wide frames (e.g. biobank: 22 cols × 194k
2219    // rows) the serial outer loop dominated ingest time, so fan the per-column
2220    // inference passes out over rayon. Order is preserved because `map` over an
2221    // indexed parallel iterator collects back in column order.
2222    let schema_cols = headers
2223        .par_iter()
2224        .enumerate()
2225        .map(|(j, name)| infer_schema_column(name, &records, j).map_err(String::from))
2226        .collect::<Result<Vec<SchemaColumn>, String>>()?;
2227    let schema = DataSchema {
2228        columns: schema_cols,
2229    };
2230    encode_recordswith_schema(headers, records, &schema, UnseenCategoryPolicy::Error)
2231}
2232
2233pub fn encode_recordswith_schema(
2234    headers: Vec<String>,
2235    records: Vec<StringRecord>,
2236    schema: &DataSchema,
2237    unseen_policy: UnseenCategoryPolicy,
2238) -> Result<EncodedDataset, String> {
2239    let n = records.len();
2240    if n == 0 {
2241        return Err(DataError::EmptyInput {
2242            reason: "table data cannot be empty".to_string(),
2243        }
2244        .into());
2245    }
2246    let p = headers.len();
2247    if p == 0 {
2248        return Err(DataError::EmptyInput {
2249            reason: "table data must have at least one header column".to_string(),
2250        }
2251        .into());
2252    }
2253    // Validate the row-width invariant up front. Without this check, records
2254    // wider than `headers` would be silently truncated (only the first
2255    // `headers.len()` fields per record would be encoded) and records
2256    // narrower than `headers` would only fail late when a per-column
2257    // `rec.get(j)` lookup returned `None`. Reject both cases explicitly so
2258    // callers cannot accidentally drop data via header/record shape skew.
2259    for (i, rec) in records.iter().enumerate() {
2260        if rec.len() != p {
2261            return Err(DataError::SchemaMismatch {
2262                reason: format!(
2263                    "row width mismatch at row {}: got {} fields, expected {} (one per header)",
2264                    i + 1,
2265                    rec.len(),
2266                    p
2267                ),
2268            }
2269            .into());
2270        }
2271    }
2272    let schema_byname: HashMap<&str, &SchemaColumn> = schema
2273        .columns
2274        .iter()
2275        .map(|c| (c.name.as_str(), c))
2276        .collect();
2277
2278    // Each column is encoded independently from the same row-major records, so
2279    // fan the per-column passes out over rayon (columns, not rows, so threads
2280    // never contend on a shared output cell). Each task returns its dense
2281    // `(kind, Vec<f64>)`; we then assemble the row-major `Array2` from the
2282    // collected columns. For wide frames this is the dominant ingest cost.
2283    let encoded_columns = headers
2284        .par_iter()
2285        .enumerate()
2286        .map(|(j, name)| {
2287            let inferred_for_extra;
2288            let col_schema = if let Some(s) = schema_byname.get(name.as_str()) {
2289                *s
2290            } else {
2291                inferred_for_extra =
2292                    infer_schema_column(name, &records, j).map_err(String::from)?;
2293                &inferred_for_extra
2294            };
2295            let column = encode_one_column(name, &records, j, col_schema, &unseen_policy)?;
2296            Ok::<(ColumnKindTag, Vec<f64>), String>((col_schema.kind, column))
2297        })
2298        .collect::<Result<Vec<(ColumnKindTag, Vec<f64>)>, String>>()?;
2299
2300    let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2301    let mut values = Array2::<f64>::zeros((n, p));
2302    for (j, (kind, column)) in encoded_columns.into_iter().enumerate() {
2303        column_kinds.push(kind);
2304        values
2305            .column_mut(j)
2306            .assign(&ndarray::ArrayView1::from(&column));
2307    }
2308
2309    Ok(EncodedDataset {
2310        headers,
2311        values,
2312        schema: schema.clone(),
2313        column_kinds,
2314    })
2315}
2316
2317/// Encode a single column `j` of `records` to its dense `f64` representation
2318/// under `col_schema`. Continuous/binary values are parsed; categorical values
2319/// are mapped to their level index (or the unseen code under `unseen_policy`).
2320/// This is the per-column work unit fanned out across columns in
2321/// [`encode_recordswith_schema`]; it scans only field `j` of each record so
2322/// distinct columns never touch shared state.
2323fn encode_one_column(
2324    name: &str,
2325    records: &[StringRecord],
2326    j: usize,
2327    col_schema: &SchemaColumn,
2328    unseen_policy: &UnseenCategoryPolicy,
2329) -> Result<Vec<f64>, String> {
2330    let level_map = if matches!(col_schema.kind, ColumnKindTag::Categorical) {
2331        Some(
2332            col_schema
2333                .levels
2334                .iter()
2335                .enumerate()
2336                .map(|(idx, v)| (v.as_str(), idx as f64))
2337                .collect::<HashMap<_, _>>(),
2338        )
2339    } else {
2340        None
2341    };
2342
2343    let mut column = Vec::<f64>::with_capacity(records.len());
2344    for (i, rec) in records.iter().enumerate() {
2345        let raw = rec
2346            .get(j)
2347            .ok_or_else(|| {
2348                String::from(DataError::SchemaMismatch {
2349                    reason: format!("missing field at row {}, col {}", i + 1, j + 1),
2350                })
2351            })?
2352            .trim();
2353        if raw.is_empty() {
2354            return Err(DataError::EmptyInput {
2355                reason: format!("empty field at row {}, column '{}'", i + 1, name),
2356            }
2357            .into());
2358        }
2359        let val = match col_schema.kind {
2360            // The missing marker is the missing value, not a parse failure (#2495).
2361            ColumnKindTag::Continuous if is_missing_marker(raw) => f64::NAN,
2362            ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
2363                String::from(DataError::SchemaMismatch {
2364                    reason: format!(
2365                        "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
2366                        name,
2367                        i + 1,
2368                        raw,
2369                        err
2370                    ),
2371                })
2372            })?,
2373            ColumnKindTag::Binary if is_missing_marker(raw) => f64::NAN,
2374            ColumnKindTag::Binary => {
2375                let v = raw.parse::<f64>().map_err(|err| {
2376                    String::from(DataError::SchemaMismatch {
2377                        reason: format!(
2378                            "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
2379                            name,
2380                            i + 1,
2381                            raw,
2382                            err
2383                        ),
2384                    })
2385                })?;
2386                if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2387                    return Err(DataError::SchemaMismatch {
2388                        reason: format!(
2389                            "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2390                            name,
2391                            i + 1,
2392                            v
2393                        ),
2394                    }
2395                    .into());
2396                }
2397                v
2398            }
2399            ColumnKindTag::Categorical => {
2400                let map = level_map.as_ref().ok_or_else(|| {
2401                    String::from(DataError::EncodingFailure {
2402                        reason: "internal categorical schema map missing".to_string(),
2403                    })
2404                })?;
2405                match map.get(raw) {
2406                    Some(v) => *v,
2407                    None => unseen_policy
2408                        .unseen_code_for(name, col_schema.levels.len())
2409                        .ok_or_else(|| {
2410                            String::from(DataError::SchemaMismatch {
2411                                reason: format!(
2412                                    "unseen level '{}' in categorical column '{}' at row {}; allowed levels: {}",
2413                                    raw,
2414                                    name,
2415                                    i + 1,
2416                                    col_schema.levels.join(",")
2417                                ),
2418                            })
2419                        })?,
2420                }
2421            }
2422        };
2423        // NaN here is the encoded missing marker (#2495), not a bad value.
2424        if !val.is_finite() && !is_missing_marker(raw) {
2425            return Err(DataError::InvalidValue {
2426                reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2427            }
2428            .into());
2429        }
2430        column.push(val);
2431    }
2432    Ok(column)
2433}
2434
2435fn infer_schema_column(
2436    name: &str,
2437    records: &[StringRecord],
2438    col_idx: usize,
2439) -> Result<SchemaColumn, DataError> {
2440    let mut all_numeric = true;
2441    let mut all_binary = true;
2442    let mut saw_numeric = false;
2443    let mut levels = Vec::<String>::new();
2444    let mut level_index = HashMap::<String, usize>::new();
2445    // Cells that are missing markers, held aside during the scan. Whether they
2446    // are levels or missing values is a property of the COLUMN, not the cell,
2447    // and is only decidable once every cell has been seen — see below.
2448    let mut missing_markers = Vec::<String>::new();
2449    for (i, rec) in records.iter().enumerate() {
2450        let raw = rec
2451            .get(col_idx)
2452            .ok_or_else(|| DataError::SchemaMismatch {
2453                reason: format!("missing field at row {}, col {}", i + 1, col_idx + 1),
2454            })?
2455            .trim();
2456        if raw.is_empty() {
2457            return Err(DataError::EmptyInput {
2458                reason: format!("empty field at row {}, column '{}'", i + 1, name),
2459            });
2460        }
2461        if is_missing_marker(raw) {
2462            missing_markers.push(raw.to_string());
2463            continue;
2464        }
2465        if let Ok(v) = raw.parse::<f64>() {
2466            saw_numeric = true;
2467            if !v.is_finite() {
2468                return Err(DataError::InvalidValue {
2469                    reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2470                });
2471            }
2472            if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2473                all_binary = false;
2474            }
2475        } else {
2476            all_numeric = false;
2477            all_binary = false;
2478            level_index.entry(raw.to_string()).or_insert_with(|| {
2479                let idx = levels.len();
2480                levels.push(raw.to_string());
2481                idx
2482            });
2483        }
2484    }
2485    // #2495: a column whose present cells are all numeric is a NUMERIC column
2486    // with missing values. It must never be re-typed categorical just because
2487    // some cells are absent — that turned every distinct measurement into a
2488    // factor level and handed the level INDEX back as the value.
2489    //
2490    // When the column is genuinely categorical (it has non-numeric cells that
2491    // are not missing markers) the markers stay levels, exactly as before. That
2492    // is deliberate: `NA` is a real category label in the wild (it is Namibia's
2493    // ISO code), and only the numeric case is unambiguous enough to reinterpret.
2494    // `saw_numeric` guards the all-missing column: with no present cell at all
2495    // there is no evidence it is numeric, so it stays categorical (one degenerate
2496    // level) rather than silently becoming an all-NaN Binary column.
2497    let numeric_column = all_numeric && saw_numeric;
2498    if !numeric_column {
2499        for marker in missing_markers {
2500            level_index.entry(marker.clone()).or_insert_with(|| {
2501                let idx = levels.len();
2502                levels.push(marker);
2503                idx
2504            });
2505        }
2506    }
2507    let kind = if numeric_column {
2508        if all_binary {
2509            ColumnKindTag::Binary
2510        } else {
2511            ColumnKindTag::Continuous
2512        }
2513    } else {
2514        ColumnKindTag::Categorical
2515    };
2516    // Canonical natural-sorted level order — see `infer_and_encode_column_major`. The
2517    // record-driven and column-major inference paths must produce byte-identical
2518    // schemas, so both sort the level set with the same natural comparator
2519    // (#1319).
2520    if matches!(kind, ColumnKindTag::Categorical) {
2521        sort_levels_canonical(&mut levels);
2522    }
2523    Ok(SchemaColumn {
2524        name: name.to_string(),
2525        kind,
2526        levels: if matches!(kind, ColumnKindTag::Categorical) {
2527            levels
2528        } else {
2529            Vec::new()
2530        },
2531    })
2532}
2533
2534/// Infer the schema of, and densely encode, a single column presented in
2535/// column-major form (`name` + its raw string field for every row).
2536///
2537/// This is the column-major sibling of the record-driven path: it produces the
2538/// byte-identical `(SchemaColumn, Vec<f64>)` that `encode_recordswith_inferred_schema`
2539/// would produce for the same column, but it reads from a `&[&str]` column
2540/// slice instead of indexing field `col_idx` of every `StringRecord`. It exists
2541/// so callers holding column-major data (e.g. the Python FFI, which can
2542/// fingerprint and cache invariant columns shared across many fits of the same
2543/// base cohort) can encode one column at a time without first materializing the
2544/// full row-major record table. `col_index` is 1-based only for error text and
2545/// matches the record-driven messages.
2546pub fn infer_and_encode_column_major(
2547    name: &str,
2548    column: &[&str],
2549    col_index: usize,
2550) -> Result<(SchemaColumn, Vec<f64>), String> {
2551    if column.is_empty() {
2552        return Err(DataError::EmptyInput {
2553            reason: "table data cannot be empty".to_string(),
2554        }
2555        .into());
2556    }
2557    // A typed Python frame prefixes every cell of a categorical-dtype column
2558    // with `CATEGORICAL_CELL_SENTINEL` so the column is encoded as a factor even
2559    // when its labels parse as numbers ("0","1","2"). Detect and strip the
2560    // marker before inference; its presence forces `Categorical` (#1317/#1318).
2561    let force_categorical = column.iter().any(|c| strip_categorical_sentinel(c).1);
2562    let mut all_numeric = !force_categorical;
2563    let mut all_binary = !force_categorical;
2564    let mut levels = Vec::<String>::new();
2565    let mut level_index = HashMap::<String, usize>::new();
2566    let mut trimmed = Vec::<&str>::with_capacity(column.len());
2567    // Capture the parsed numeric value alongside each trimmed field during the
2568    // single inference scan, so the encode pass below never re-parses a numeric
2569    // string. For wide biobank frames the f64 parse dominated ingest, and the
2570    // record-driven path used to parse every continuous/binary field twice
2571    // (once to infer the schema, once to encode). `parsed[i]` is `Some(v)` iff
2572    // field `i` parsed as a finite f64; categorical columns ignore it.
2573    let mut parsed = Vec::<Option<f64>>::with_capacity(column.len());
2574    let mut saw_numeric = false;
2575    let mut missing_positions = Vec::<usize>::new();
2576    for (i, raw_field) in column.iter().enumerate() {
2577        // Strip the categorical marker (if any) so the recorded level label and
2578        // any numeric parse see the user's clean text, not the sentinel.
2579        let (raw, _) = strip_categorical_sentinel(raw_field);
2580        let raw = raw.trim();
2581        if raw.is_empty() {
2582            return Err(DataError::EmptyInput {
2583                reason: format!("empty field at row {}, column '{}'", i + 1, name),
2584            }
2585            .into());
2586        }
2587        // When the source column is dtype-categorical, every cell is a level
2588        // regardless of whether its label parses as a number.
2589        if !force_categorical {
2590            // #2495: hold missing markers aside without disturbing the numeric
2591            // verdict. Whether they are levels or missing values depends on the
2592            // whole column, so it is resolved after the scan.
2593            if is_missing_marker(raw) {
2594                missing_positions.push(i);
2595                parsed.push(Some(f64::NAN));
2596                trimmed.push(raw);
2597                continue;
2598            }
2599            if let Ok(v) = raw.parse::<f64>() {
2600                saw_numeric = true;
2601                if !v.is_finite() {
2602                    return Err(DataError::InvalidValue {
2603                        reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2604                    }
2605                    .into());
2606                }
2607                if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2608                    all_binary = false;
2609                }
2610                parsed.push(Some(v));
2611                trimmed.push(raw);
2612                continue;
2613            }
2614            all_numeric = false;
2615            all_binary = false;
2616        }
2617        level_index.entry(raw.to_string()).or_insert_with(|| {
2618            let idx = levels.len();
2619            levels.push(raw.to_string());
2620            idx
2621        });
2622        parsed.push(None);
2623        trimmed.push(raw);
2624    }
2625    // #2495, mirroring `infer_schema_column`: present cells all numeric => a
2626    // numeric column with missing values, never a factor over its own
2627    // measurements. Otherwise the markers are ordinary levels (`NA` is a real
2628    // label in the wild), so put them back where the scan would have.
2629    let numeric_column = all_numeric && saw_numeric;
2630    if !numeric_column {
2631        for &i in &missing_positions {
2632            let raw = trimmed[i];
2633            level_index.entry(raw.to_string()).or_insert_with(|| {
2634                let idx = levels.len();
2635                levels.push(raw.to_string());
2636                idx
2637            });
2638            parsed[i] = None;
2639        }
2640    }
2641    let kind = if numeric_column {
2642        if all_binary {
2643            ColumnKindTag::Binary
2644        } else {
2645            ColumnKindTag::Continuous
2646        }
2647    } else {
2648        ColumnKindTag::Categorical
2649    };
2650    // Canonical level ordering: sort factor levels lexicographically rather than
2651    // recording them in first-appearance order. Every reference tool a gam user
2652    // comes from — R `factor()` (C-locale sort), pandas `Categorical`, sklearn
2653    // `LabelEncoder` — orders categorical levels canonically, and downstream
2654    // consumers key off that order: the multinomial driver lays out one output
2655    // probability column per level and takes the *last* level as the softmax
2656    // reference, so first-appearance order made the `(n, K)` prediction columns
2657    // depend on which class happened to appear first in the training rows (a
2658    // row-shuffle would permute the output) instead of on the class labels
2659    // (#1319). Sorting makes the encoding a deterministic function of the label
2660    // *set*, independent of row order, and matches the factor convention so
2661    // column `k` of a multinomial prediction is class `levels[k]`. Use natural
2662    // ordering so generated labels like g2 stay before g10.
2663    if matches!(kind, ColumnKindTag::Categorical) {
2664        sort_levels_canonical(&mut levels);
2665    }
2666    let schema = SchemaColumn {
2667        name: name.to_string(),
2668        kind,
2669        levels: if matches!(kind, ColumnKindTag::Categorical) {
2670            levels
2671        } else {
2672            Vec::new()
2673        },
2674    };
2675
2676    let level_map = if matches!(kind, ColumnKindTag::Categorical) {
2677        Some(
2678            schema
2679                .levels
2680                .iter()
2681                .enumerate()
2682                .map(|(idx, v)| (v.as_str(), idx as f64))
2683                .collect::<HashMap<_, _>>(),
2684        )
2685    } else {
2686        None
2687    };
2688
2689    let mut values = Vec::<f64>::with_capacity(trimmed.len());
2690    for (i, raw) in trimmed.iter().enumerate() {
2691        let raw = *raw;
2692        let val = match kind {
2693            // Continuous/Binary kinds are only selected when every field parsed
2694            // as a finite f64 during inference, so `parsed[i]` is always `Some`
2695            // here — reuse it instead of re-parsing the string.
2696            ColumnKindTag::Continuous => parsed[i].ok_or_else(|| {
2697                String::from(DataError::EncodingFailure {
2698                    reason: format!(
2699                        "internal: continuous column '{}' lost its parsed value at row {} (col {})",
2700                        name,
2701                        i + 1,
2702                        col_index
2703                    ),
2704                })
2705            })?,
2706            ColumnKindTag::Binary => {
2707                let v = parsed[i].ok_or_else(|| {
2708                    String::from(DataError::EncodingFailure {
2709                        reason: format!(
2710                            "internal: binary column '{}' lost its parsed value at row {} (col {})",
2711                            name,
2712                            i + 1,
2713                            col_index
2714                        ),
2715                    })
2716                })?;
2717                // A missing cell carries NaN (#2495); that is absence, not a
2718                // 0/1 violation.
2719                if v.is_finite() && (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2720                    return Err(DataError::SchemaMismatch {
2721                        reason: format!(
2722                            "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2723                            name,
2724                            i + 1,
2725                            v
2726                        ),
2727                    }
2728                    .into());
2729                }
2730                v
2731            }
2732            ColumnKindTag::Categorical => {
2733                let map = level_map.as_ref().ok_or_else(|| {
2734                    String::from(DataError::EncodingFailure {
2735                        reason: "internal categorical schema map missing".to_string(),
2736                    })
2737                })?;
2738                *map.get(raw).ok_or_else(|| {
2739                    String::from(DataError::EncodingFailure {
2740                        reason: format!(
2741                            "internal: level '{}' missing from freshly built map for column '{}' (col {})",
2742                            raw, name, col_index
2743                        ),
2744                    })
2745                })?
2746            }
2747        };
2748        // NaN here is the encoded missing marker (#2495), not a bad value.
2749        if !val.is_finite() && !is_missing_marker(raw) {
2750            return Err(DataError::InvalidValue {
2751                reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2752            }
2753            .into());
2754        }
2755        values.push(val);
2756    }
2757    Ok((schema, values))
2758}
2759
2760#[cfg(test)]
2761mod missing_value_inference_tests {
2762    use super::*;
2763
2764    fn rows(cells: &[&[&str]]) -> Vec<StringRecord> {
2765        cells
2766            .iter()
2767            .map(|r| StringRecord::from(r.to_vec()))
2768            .collect()
2769    }
2770
2771    /// The #2495 defect, pinned. `parker` in `bench/datasets/wine.csv` is 29
2772    /// Parker scores spanning 65.0–94.4 plus 18 `NA`s. Before the fix the single
2773    /// non-parsing token re-typed the whole column CATEGORICAL, every distinct
2774    /// score became a factor level, and `encode(raw) as f64` handed back the
2775    /// level INDEX — so a 65–94 measurement arrived downstream as 0…29 and
2776    /// `is_finite()` stopped being an NA test.
2777    #[test]
2778    fn a_numeric_column_containing_na_can_never_infer_categorical() {
2779        let ds = encode_recordswith_inferred_schema(
2780            vec!["parker".to_string()],
2781            rows(&[&["94.4"], &["NA"], &["65.0"], &["88.0"], &["NA"]]),
2782        )
2783        .expect("encode a numeric column carrying NA");
2784
2785        assert_eq!(
2786            ds.schema.columns[0].kind,
2787            ColumnKindTag::Continuous,
2788            "a column whose present cells are all numeric is numeric-with-missing, \
2789             never a factor over its own measurements"
2790        );
2791        assert!(
2792            ds.schema.columns[0].levels.is_empty(),
2793            "measurements must not be recorded as factor levels"
2794        );
2795
2796        // The values survive as themselves, and the missing cells are NaN — the
2797        // representation callers already assume when they filter on is_finite().
2798        let col: Vec<f64> = ds.values.column(0).to_vec();
2799        assert_eq!(col[0], 94.4);
2800        assert_eq!(col[2], 65.0);
2801        assert_eq!(col[3], 88.0);
2802        assert!(col[1].is_nan() && col[4].is_nan(), "NA must encode as NaN");
2803        assert_eq!(
2804            col.iter().filter(|v| v.is_finite()).count(),
2805            3,
2806            "is_finite() must count exactly the present cells"
2807        );
2808    }
2809
2810    /// The other half, and the reason the rule is about the COLUMN and not the
2811    /// cell: `NA` is a real category label in the wild (Namibia's ISO code). In a
2812    /// genuinely categorical column it must stay an ordinary level, not silently
2813    /// become missing data.
2814    #[test]
2815    fn na_stays_a_level_in_a_genuinely_categorical_column() {
2816        let ds = encode_recordswith_inferred_schema(
2817            vec!["country".to_string()],
2818            rows(&[&["NA"], &["ZA"], &["BW"], &["NA"]]),
2819        )
2820        .expect("encode a categorical column whose labels include NA");
2821
2822        assert_eq!(ds.schema.columns[0].kind, ColumnKindTag::Categorical);
2823        assert!(
2824            ds.schema.columns[0].levels.iter().any(|l| l == "NA"),
2825            "NA is a country here, not missingness: levels were {:?}",
2826            ds.schema.columns[0].levels
2827        );
2828        assert!(
2829            ds.values.column(0).iter().all(|v| v.is_finite()),
2830            "a categorical column carries level codes, never NaN"
2831        );
2832    }
2833
2834    /// Binary columns take the same rule: 0/1 with holes stays binary, and the
2835    /// hole is not a 0/1 violation.
2836    #[test]
2837    fn a_binary_column_containing_na_stays_binary_with_nan_holes() {
2838        let ds = encode_recordswith_inferred_schema(
2839            vec!["event".to_string()],
2840            rows(&[&["1"], &["NA"], &["0"], &["1"]]),
2841        )
2842        .expect("encode a binary column carrying NA");
2843
2844        assert_eq!(ds.schema.columns[0].kind, ColumnKindTag::Binary);
2845        let col: Vec<f64> = ds.values.column(0).to_vec();
2846        assert_eq!((col[0], col[2], col[3]), (1.0, 0.0, 1.0));
2847        assert!(col[1].is_nan());
2848    }
2849
2850    /// `NaN` and `inf` DO parse via `f64::from_str`, so they are not missing
2851    /// markers — they hit the existing non-finite guard and fail loudly. That
2852    /// distinction is the whole point: the silent class is the tokens that
2853    /// neither parse as a number nor name a category.
2854    #[test]
2855    fn a_parsed_non_finite_literal_still_fails_loudly() {
2856        let err = encode_recordswith_inferred_schema(
2857            vec!["x".to_string()],
2858            rows(&[&["1.0"], &["inf"], &["2.0"]]),
2859        )
2860        .expect_err("a literal infinity is a data error, not a missing value");
2861        assert!(
2862            err.contains("non-finite"),
2863            "expected the non-finite guard, got: {err}"
2864        );
2865    }
2866}
2867
2868#[cfg(test)]
2869mod tests {
2870    use super::*;
2871    use arrow::array::ArrayRef;
2872    use arrow::datatypes::{Field, Schema};
2873    use arrow::error::ArrowError;
2874    use arrow::record_batch::{RecordBatch, RecordBatchIterator};
2875    use std::sync::Arc;
2876
2877    fn encode_single_arrow_array(array: ArrayRef) -> Result<EncodedDataset, DataError> {
2878        let schema = Arc::new(Schema::new(vec![Field::new(
2879            "source",
2880            array.data_type().clone(),
2881            true,
2882        )]));
2883        let batch = RecordBatch::try_new(schema.clone(), vec![array]).expect("record batch");
2884        let batches: Vec<Result<RecordBatch, ArrowError>> = vec![Ok(batch)];
2885        let mut reader = RecordBatchIterator::new(batches, schema);
2886        encode_arrow_record_batch_reader_with_inferred_schema(
2887            &mut reader,
2888            vec!["normalized".to_string()],
2889        )
2890    }
2891
2892    #[test]
2893    fn arrow_reader_streams_typed_columns_in_supplied_order() {
2894        use arrow::array::{
2895            Array, BooleanArray, DictionaryArray, Float32Array, Int8Array, Int32Array, Int64Array,
2896            LargeStringArray, StringArray,
2897        };
2898        use arrow::datatypes::Int8Type;
2899
2900        let string_dictionary_1: DictionaryArray<Int8Type> =
2901            vec!["beta", "alpha"].into_iter().collect();
2902        let string_dictionary_2: DictionaryArray<Int8Type> = vec!["beta"].into_iter().collect();
2903        let numeric_dictionary_1 = DictionaryArray::<Int8Type>::new(
2904            Int8Array::from(vec![0, 1]),
2905            Arc::new(Int64Array::from(vec![5, 7])),
2906        );
2907        let numeric_dictionary_2 = DictionaryArray::<Int8Type>::new(
2908            Int8Array::from(vec![0]),
2909            Arc::new(Int64Array::from(vec![5])),
2910        );
2911
2912        let schema = Arc::new(Schema::new(vec![
2913            Field::new("source_float", arrow::datatypes::DataType::Float32, false),
2914            Field::new("source_integer", arrow::datatypes::DataType::Int32, false),
2915            Field::new("source_flag", arrow::datatypes::DataType::Boolean, false),
2916            Field::new("source_utf8", arrow::datatypes::DataType::Utf8, false),
2917            Field::new(
2918                "source_large_utf8",
2919                arrow::datatypes::DataType::LargeUtf8,
2920                false,
2921            ),
2922            Field::new(
2923                "source_dictionary_string",
2924                string_dictionary_1.data_type().clone(),
2925                false,
2926            ),
2927            Field::new(
2928                "source_dictionary_number",
2929                numeric_dictionary_1.data_type().clone(),
2930                false,
2931            ),
2932        ]));
2933        let batch_1 = RecordBatch::try_new(
2934            schema.clone(),
2935            vec![
2936                Arc::new(Float32Array::from(vec![1.5, 2.5])) as ArrayRef,
2937                Arc::new(Int32Array::from(vec![0, 1])),
2938                Arc::new(BooleanArray::from(vec![true, false])),
2939                Arc::new(StringArray::from(vec!["item10", "item2"])),
2940                Arc::new(LargeStringArray::from(vec!["z", "a"])),
2941                Arc::new(string_dictionary_1),
2942                Arc::new(numeric_dictionary_1),
2943            ],
2944        )
2945        .expect("first record batch");
2946        let batch_2 = RecordBatch::try_new(
2947            schema.clone(),
2948            vec![
2949                Arc::new(Float32Array::from(vec![-4.0])) as ArrayRef,
2950                Arc::new(Int32Array::from(vec![1])),
2951                Arc::new(BooleanArray::from(vec![true])),
2952                Arc::new(StringArray::from(vec!["item1"])),
2953                Arc::new(LargeStringArray::from(vec!["z"])),
2954                Arc::new(string_dictionary_2),
2955                Arc::new(numeric_dictionary_2),
2956            ],
2957        )
2958        .expect("second record batch");
2959        let batches: Vec<Result<RecordBatch, ArrowError>> = vec![Ok(batch_1), Ok(batch_2)];
2960        let mut reader = RecordBatchIterator::new(batches, schema);
2961        let headers = [
2962            "float",
2963            "integer",
2964            "flag",
2965            "utf8",
2966            "large_utf8",
2967            "dictionary_string",
2968            "dictionary_number",
2969        ]
2970        .map(str::to_string)
2971        .to_vec();
2972
2973        let dataset =
2974            encode_arrow_record_batch_reader_with_inferred_schema(&mut reader, headers.clone())
2975                .expect("Arrow stream should encode");
2976
2977        assert_eq!(dataset.headers, headers);
2978        assert_eq!(
2979            dataset.column_kinds,
2980            vec![
2981                ColumnKindTag::Continuous,
2982                ColumnKindTag::Binary,
2983                ColumnKindTag::Binary,
2984                ColumnKindTag::Categorical,
2985                ColumnKindTag::Categorical,
2986                ColumnKindTag::Categorical,
2987                ColumnKindTag::Continuous,
2988            ]
2989        );
2990        assert_eq!(
2991            dataset.values,
2992            ndarray::arr2(&[
2993                [1.5, 0.0, 1.0, 2.0, 1.0, 1.0, 5.0],
2994                [2.5, 1.0, 0.0, 1.0, 0.0, 0.0, 7.0],
2995                [-4.0, 1.0, 1.0, 0.0, 1.0, 1.0, 5.0],
2996            ])
2997        );
2998        assert_eq!(
2999            dataset.schema.columns[3].levels,
3000            vec!["item1", "item2", "item10"]
3001        );
3002        assert_eq!(dataset.schema.columns[4].levels, vec!["a", "z"]);
3003        assert_eq!(dataset.schema.columns[5].levels, vec!["alpha", "beta"]);
3004        assert!(
3005            dataset
3006                .schema
3007                .columns
3008                .iter()
3009                .zip(dataset.headers.iter())
3010                .all(|(column, header)| column.name == *header)
3011        );
3012    }
3013
3014    #[test]
3015    fn arrow_reader_rejects_empty_duplicate_and_mismatched_headers() {
3016        let schema = Arc::new(Schema::new(vec![Field::new(
3017            "source",
3018            arrow::datatypes::DataType::Int32,
3019            false,
3020        )]));
3021
3022        let mut empty_name_reader = RecordBatchIterator::new(
3023            Vec::<Result<RecordBatch, ArrowError>>::new(),
3024            schema.clone(),
3025        );
3026        let empty_name = encode_arrow_record_batch_reader_with_inferred_schema(
3027            &mut empty_name_reader,
3028            vec!["  ".to_string()],
3029        )
3030        .expect_err("blank header should fail");
3031        assert!(matches!(empty_name, DataError::EmptyInput { .. }));
3032
3033        let mut duplicate_reader = RecordBatchIterator::new(
3034            Vec::<Result<RecordBatch, ArrowError>>::new(),
3035            Arc::new(Schema::new(vec![
3036                Field::new("a", arrow::datatypes::DataType::Int32, false),
3037                Field::new("b", arrow::datatypes::DataType::Int32, false),
3038            ])),
3039        );
3040        let duplicate = encode_arrow_record_batch_reader_with_inferred_schema(
3041            &mut duplicate_reader,
3042            vec!["x".to_string(), "x".to_string()],
3043        )
3044        .expect_err("duplicate header should fail");
3045        assert!(matches!(duplicate, DataError::SchemaMismatch { .. }));
3046
3047        let mut mismatch_reader =
3048            RecordBatchIterator::new(Vec::<Result<RecordBatch, ArrowError>>::new(), schema);
3049        let mismatch = encode_arrow_record_batch_reader_with_inferred_schema(
3050            &mut mismatch_reader,
3051            vec!["x".to_string(), "y".to_string()],
3052        )
3053        .expect_err("header count mismatch should fail");
3054        assert!(matches!(mismatch, DataError::SchemaMismatch { .. }));
3055    }
3056
3057    #[test]
3058    fn arrow_reader_reports_typed_null_nonfinite_and_unsupported_errors() {
3059        use arrow::array::{Date32Array, DictionaryArray, Float64Array, Int8Array, StringArray};
3060        use arrow::datatypes::Int8Type;
3061
3062        let null_numeric =
3063            encode_single_arrow_array(Arc::new(Float64Array::from(vec![Some(1.0), None])))
3064                .expect_err("numeric null should fail");
3065        assert!(matches!(&null_numeric, DataError::InvalidValue { .. }));
3066        assert!(null_numeric.to_string().contains("null value at row 2"));
3067
3068        let null_dictionary = DictionaryArray::<Int8Type>::new(
3069            Int8Array::from(vec![0, 1]),
3070            Arc::new(StringArray::from(vec![Some("present"), None])),
3071        );
3072        let logical_null = encode_single_arrow_array(Arc::new(null_dictionary))
3073            .expect_err("null dictionary value should fail");
3074        assert!(matches!(&logical_null, DataError::InvalidValue { .. }));
3075        assert!(logical_null.to_string().contains("null value at row 2"));
3076
3077        let nonfinite = encode_single_arrow_array(Arc::new(Float64Array::from(vec![f64::NAN])))
3078            .expect_err("NaN should fail");
3079        assert!(matches!(&nonfinite, DataError::InvalidValue { .. }));
3080        assert!(nonfinite.to_string().contains("non-finite value"));
3081
3082        let unsupported = encode_single_arrow_array(Arc::new(Date32Array::from(vec![1])))
3083            .expect_err("date column should fail");
3084        assert!(matches!(&unsupported, DataError::InvalidValue { .. }));
3085        assert!(
3086            unsupported
3087                .to_string()
3088                .contains("unsupported Arrow column type")
3089        );
3090    }
3091
3092    #[test]
3093    fn encode_records_rejects_empty_input() {
3094        let headers = vec!["x".to_string()];
3095        let schema = DataSchema {
3096            columns: vec![SchemaColumn {
3097                name: "x".to_string(),
3098                kind: ColumnKindTag::Continuous,
3099                levels: Vec::new(),
3100            }],
3101        };
3102
3103        let err = encode_recordswith_inferred_schema(headers.clone(), Vec::new())
3104            .expect_err("empty inferred records should error");
3105        assert_eq!(err, "table data cannot be empty");
3106
3107        let err =
3108            encode_recordswith_schema(headers, Vec::new(), &schema, UnseenCategoryPolicy::Error)
3109                .expect_err("empty schema-guided records should error");
3110        assert_eq!(err, "table data cannot be empty");
3111    }
3112
3113    #[test]
3114    fn column_major_matches_record_driven_inferred_encode() {
3115        // The FFI ingest path encodes column-by-column via
3116        // `infer_and_encode_column_major`; it must produce byte-identical
3117        // schema + values to the record-driven `encode_recordswith_inferred_schema`
3118        // for the same frame across all three inferred kinds.
3119        let headers = vec!["cont".to_string(), "bin".to_string(), "cat".to_string()];
3120        let raw_rows = vec![
3121            vec!["1.5", "0", "a"],
3122            vec!["2.0", "1", "b"],
3123            vec!["-3.25", "1", "a"],
3124            vec!["0.0", "0", "c"],
3125        ];
3126        let records: Vec<StringRecord> = raw_rows
3127            .iter()
3128            .map(|r| StringRecord::from(r.clone()))
3129            .collect();
3130        let record_ds = encode_recordswith_inferred_schema(headers.clone(), records)
3131            .expect("record-driven encode");
3132
3133        for (j, name) in headers.iter().enumerate() {
3134            let column: Vec<&str> = raw_rows.iter().map(|r| r[j]).collect();
3135            let (schema_col, values) =
3136                infer_and_encode_column_major(name, &column, j + 1).expect("column-major encode");
3137            assert_eq!(schema_col.kind, record_ds.schema.columns[j].kind);
3138            assert_eq!(schema_col.levels, record_ds.schema.columns[j].levels);
3139            for (i, v) in values.iter().enumerate() {
3140                assert_eq!(*v, record_ds.values[[i, j]], "row {i} col {name}");
3141            }
3142        }
3143    }
3144
3145    #[test]
3146    fn encode_records_can_encode_unseen_named_categorical_column() {
3147        let schema = DataSchema {
3148            columns: vec![
3149                SchemaColumn {
3150                    name: "g".to_string(),
3151                    kind: ColumnKindTag::Categorical,
3152                    levels: vec!["a".to_string(), "b".to_string()],
3153                },
3154                SchemaColumn {
3155                    name: "x".to_string(),
3156                    kind: ColumnKindTag::Categorical,
3157                    levels: vec!["low".to_string(), "high".to_string()],
3158                },
3159            ],
3160        };
3161        let headers = vec!["g".to_string(), "x".to_string()];
3162        let records = vec![StringRecord::from(vec!["new-group", "low"])];
3163        let policy =
3164            UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
3165
3166        let ds =
3167            encode_recordswith_schema(headers, records, &schema, policy).expect("encoded dataset");
3168
3169        assert_eq!(ds.values[[0, 0]], 2.0);
3170        assert_eq!(ds.values[[0, 1]], 0.0);
3171    }
3172
3173    #[test]
3174    fn categorical_encoder_consumes_labels_and_remaps_canonically() {
3175        use ndarray::Array1;
3176
3177        let mut encoder = CategoricalEncoder::default();
3178        let mut encoded = Array1::from_vec(
3179            ["item10", "item2", "item1", "item2"]
3180                .into_iter()
3181                .map(|label| encoder.encode(label) as f64)
3182                .collect(),
3183        );
3184
3185        let levels = encoder.finish(encoded.view_mut(), LevelOrder::Canonical);
3186
3187        assert_eq!(levels, vec!["item1", "item2", "item10"]);
3188        assert_eq!(encoded.to_vec(), vec![2.0, 1.0, 0.0, 1.0]);
3189    }
3190
3191    #[test]
3192    fn complete_delimited_schema_encodes_projected_rows_directly() {
3193        let dir = tempfile::tempdir().expect("tempdir");
3194        let path = dir.path().join("schema_direct.csv");
3195        std::fs::write(
3196            &path,
3197            "y,group,flag,unused\n1.5,b,0,first\n2.5,a,1,second\n",
3198        )
3199        .expect("write csv");
3200        let schema = DataSchema {
3201            columns: vec![
3202                SchemaColumn {
3203                    name: "group".to_string(),
3204                    kind: ColumnKindTag::Categorical,
3205                    levels: vec!["a".to_string(), "b".to_string()],
3206                },
3207                SchemaColumn {
3208                    name: "flag".to_string(),
3209                    kind: ColumnKindTag::Binary,
3210                    levels: Vec::new(),
3211                },
3212                SchemaColumn {
3213                    name: "y".to_string(),
3214                    kind: ColumnKindTag::Continuous,
3215                    levels: Vec::new(),
3216                },
3217            ],
3218        };
3219
3220        let loaded = load_datasetwith_schema_projected(
3221            &path,
3222            &schema,
3223            UnseenCategoryPolicy::Error,
3224            &["y".to_string(), "group".to_string(), "flag".to_string()],
3225        )
3226        .expect("schema-guided projected load");
3227
3228        assert_eq!(loaded.headers, vec!["y", "group", "flag"]);
3229        assert_eq!(loaded.values.row(0).to_vec(), vec![1.5, 1.0, 0.0]);
3230        assert_eq!(loaded.values.row(1).to_vec(), vec![2.5, 0.0, 1.0]);
3231        assert_eq!(
3232            loaded.column_kinds,
3233            vec![
3234                ColumnKindTag::Continuous,
3235                ColumnKindTag::Categorical,
3236                ColumnKindTag::Binary,
3237            ]
3238        );
3239    }
3240
3241    #[test]
3242    fn direct_parquet_decoder_preserves_string_encounter_order() {
3243        use arrow::array::DictionaryArray;
3244        use arrow::datatypes::Int8Type;
3245        use ndarray::Array1;
3246
3247        let dictionary: DictionaryArray<Int8Type> =
3248            vec!["beta", "alpha", "beta"].into_iter().collect();
3249        let mut encoded = Array1::<f64>::zeros(dictionary.len());
3250        let mut encoder = CategoricalEncoder::default();
3251        let mut all_binary = true;
3252
3253        decode_arrow_batch_column_into(
3254            &dictionary,
3255            0,
3256            "group",
3257            true,
3258            encoded.view_mut(),
3259            Some(&mut encoder),
3260            &mut all_binary,
3261        )
3262        .expect("dictionary strings decode directly");
3263        let levels = encoder.finish(encoded.view_mut(), LevelOrder::Encounter);
3264
3265        assert_eq!(levels, vec!["beta", "alpha"]);
3266        assert_eq!(encoded.to_vec(), vec![0.0, 1.0, 0.0]);
3267    }
3268
3269    #[test]
3270    fn numeric_valued_dictionary_column_classifies_and_decodes_as_numeric() {
3271        // Regression for #1162: pyarrow dictionary-encodes low-cardinality
3272        // *numeric* columns by default (e.g. `Dictionary(Int8, Int64)`).
3273        // Dictionary encoding is a storage detail, not a semantic type, so such
3274        // a column must stay numeric — both at classification time
3275        // (`arrow_field_is_string`) and at decode time
3276        // (`decode_arrow_batch_column_into`). Previously the loader matched ALL
3277        // `Dictionary(_, _)` as string/categorical, silently flipping numeric
3278        // features to categorical and rejecting valid numeric prediction files
3279        // with SchemaMismatch.
3280        use arrow::array::{Array, ArrayRef, DictionaryArray, Int8Array, Int64Array};
3281        use arrow::datatypes::{DataType, Int8Type};
3282        use std::sync::Arc;
3283
3284        // Logical column values: 5, 7, 5, 7, 5 (low-cardinality integers).
3285        let keys = Int8Array::from(vec![0i8, 1, 0, 1, 0]);
3286        let dict_values: ArrayRef = Arc::new(Int64Array::from(vec![5i64, 7]));
3287        let dict: DictionaryArray<Int8Type> = DictionaryArray::new(keys, dict_values);
3288
3289        // The dictionary's *value* type is numeric, so the column must NOT be
3290        // classified as string/categorical.
3291        assert!(matches!(dict.data_type(), DataType::Dictionary(_, _)));
3292        assert!(
3293            !arrow_field_is_string(dict.data_type()),
3294            "Dictionary(Int8, Int64) must not be treated as a string column"
3295        );
3296
3297        // A genuine string-valued dictionary still classifies as string.
3298        let str_dict: DictionaryArray<Int8Type> = vec!["a", "b", "a"].into_iter().collect();
3299        assert!(
3300            arrow_field_is_string(str_dict.data_type()),
3301            "Dictionary(Int8, Utf8) must remain a string column"
3302        );
3303
3304        // Decoding the numeric dictionary with `is_string_col = false` must
3305        // resolve indices through the dictionary and write the underlying
3306        // numeric values directly into the destination view.
3307        let mut decoded = ndarray::Array1::<f64>::zeros(dict.len());
3308        let mut all_binary = true;
3309        decode_arrow_batch_column_into(
3310            &dict,
3311            0,
3312            "x",
3313            false,
3314            decoded.view_mut(),
3315            None,
3316            &mut all_binary,
3317        )
3318        .expect("numeric dictionary column should decode as numeric");
3319        assert_eq!(decoded.to_vec(), vec![5.0, 7.0, 5.0, 7.0, 5.0]);
3320        assert!(!all_binary);
3321
3322        // End-to-end: write a real parquet file whose only column is the
3323        // dictionary-encoded numeric one, then load it both ways. This is the
3324        // exact repro from #1162 (pyarrow's default dictionary encoding of a
3325        // low-cardinality numeric column).
3326        use arrow::datatypes::{Field, Schema};
3327        use arrow::record_batch::RecordBatch;
3328        use parquet::arrow::ArrowWriter;
3329
3330        let arrow_schema = Arc::new(Schema::new(vec![Field::new(
3331            "x",
3332            dict.data_type().clone(),
3333            false,
3334        )]));
3335        let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(dict.clone())])
3336            .expect("record batch with a dictionary numeric column");
3337
3338        let dir = tempfile::tempdir().expect("tempdir");
3339        let path = dir.path().join("dict_numeric.parquet");
3340        {
3341            let file = std::fs::File::create(&path).expect("create parquet");
3342            let mut writer =
3343                ArrowWriter::try_new(file, arrow_schema, None).expect("arrow parquet writer");
3344            writer.write(&batch).expect("write batch");
3345            writer.close().expect("close writer");
3346        }
3347
3348        // Inferred load: the column must be Continuous (5 and 7 are not 0/1),
3349        // never Categorical.
3350        let inferred =
3351            load_parquet_inferred(&path, &[], &HashSet::new()).expect("inferred parquet load");
3352        assert_eq!(inferred.column_kinds, vec![ColumnKindTag::Continuous]);
3353        assert_eq!(
3354            inferred.values.column(0).to_vec(),
3355            vec![5.0, 7.0, 5.0, 7.0, 5.0]
3356        );
3357
3358        // Schema-driven load with the column declared Continuous (as it would be
3359        // after training on CSV / non-dictionary parquet) must NOT raise the
3360        // SchemaMismatch that #1162 reported on valid numeric data.
3361        let schema = DataSchema {
3362            columns: vec![SchemaColumn {
3363                name: "x".to_string(),
3364                kind: ColumnKindTag::Continuous,
3365                levels: Vec::new(),
3366            }],
3367        };
3368        let schema_loaded =
3369            load_parquet_with_schema(&path, &schema, UnseenCategoryPolicy::Error, &[])
3370                .expect("dictionary-encoded numeric parquet must load against a Continuous schema");
3371        assert_eq!(schema_loaded.column_kinds, vec![ColumnKindTag::Continuous]);
3372        assert_eq!(
3373            schema_loaded.values.column(0).to_vec(),
3374            vec![5.0, 7.0, 5.0, 7.0, 5.0]
3375        );
3376    }
3377
3378    #[test]
3379    fn encode_records_keeps_unlisted_categorical_columns_strict() {
3380        let schema = DataSchema {
3381            columns: vec![
3382                SchemaColumn {
3383                    name: "g".to_string(),
3384                    kind: ColumnKindTag::Categorical,
3385                    levels: vec!["a".to_string(), "b".to_string()],
3386                },
3387                SchemaColumn {
3388                    name: "x".to_string(),
3389                    kind: ColumnKindTag::Categorical,
3390                    levels: vec!["low".to_string(), "high".to_string()],
3391                },
3392            ],
3393        };
3394        let headers = vec!["g".to_string(), "x".to_string()];
3395        let records = vec![StringRecord::from(vec!["a", "new-level"])];
3396        let policy =
3397            UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
3398
3399        let err = encode_recordswith_schema(headers, records, &schema, policy)
3400            .expect_err("ordinary categorical column should stay strict");
3401
3402        assert!(err.contains("unseen level 'new-level' in categorical column 'x'"));
3403    }
3404
3405    // -----------------------------------------------------------------------
3406    // strip_categorical_sentinel
3407    // -----------------------------------------------------------------------
3408
3409    #[test]
3410    fn sentinel_strip_present_returns_rest_and_true() {
3411        let marked = format!("{}{}", CATEGORICAL_CELL_SENTINEL, "hello");
3412        let (rest, found) = strip_categorical_sentinel(&marked);
3413        assert_eq!(rest, "hello");
3414        assert!(found);
3415    }
3416
3417    #[test]
3418    fn sentinel_strip_absent_returns_original_and_false() {
3419        let (rest, found) = strip_categorical_sentinel("hello");
3420        assert_eq!(rest, "hello");
3421        assert!(!found);
3422    }
3423
3424    #[test]
3425    fn sentinel_strip_empty_string_returns_empty_and_false() {
3426        let (rest, found) = strip_categorical_sentinel("");
3427        assert_eq!(rest, "");
3428        assert!(!found);
3429    }
3430
3431    #[test]
3432    fn sentinel_strip_only_sentinel_returns_empty_and_true() {
3433        let marked = CATEGORICAL_CELL_SENTINEL.to_string();
3434        let (rest, found) = strip_categorical_sentinel(&marked);
3435        assert_eq!(rest, "");
3436        assert!(found);
3437    }
3438
3439    // -----------------------------------------------------------------------
3440    // EncodedDataset::feature_ranges
3441    // -----------------------------------------------------------------------
3442
3443    #[test]
3444    fn feature_ranges_two_columns() {
3445        let values = ndarray::arr2(&[[1.0_f64, 10.0], [3.0, 20.0], [2.0, 15.0]]);
3446        let ds = EncodedDataset {
3447            headers: vec!["a".to_string(), "b".to_string()],
3448            values,
3449            schema: DataSchema { columns: vec![] },
3450            column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3451        };
3452        let ranges = ds.feature_ranges();
3453        assert_eq!(ranges.len(), 2);
3454        assert_eq!(ranges[0], (1.0, 3.0));
3455        assert_eq!(ranges[1], (10.0, 20.0));
3456    }
3457
3458    #[test]
3459    fn feature_ranges_single_row_min_equals_max() {
3460        let values = ndarray::arr2(&[[5.0_f64, -3.0]]);
3461        let ds = EncodedDataset {
3462            headers: vec!["x".to_string(), "y".to_string()],
3463            values,
3464            schema: DataSchema { columns: vec![] },
3465            column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3466        };
3467        let ranges = ds.feature_ranges();
3468        assert_eq!(ranges[0], (5.0, 5.0));
3469        assert_eq!(ranges[1], (-3.0, -3.0));
3470    }
3471
3472    #[test]
3473    fn feature_ranges_all_nan_defaults_to_zero() {
3474        let values = ndarray::arr2(&[[f64::NAN], [f64::NAN]]);
3475        let ds = EncodedDataset {
3476            headers: vec!["x".to_string()],
3477            values,
3478            schema: DataSchema { columns: vec![] },
3479            column_kinds: vec![ColumnKindTag::Continuous],
3480        };
3481        let ranges = ds.feature_ranges();
3482        assert_eq!(ranges[0], (0.0, 0.0));
3483    }
3484
3485    // -----------------------------------------------------------------------
3486    // EncodedDataset::column_map
3487    // -----------------------------------------------------------------------
3488
3489    #[test]
3490    fn column_map_indexes_by_name() {
3491        let values = ndarray::arr2(&[[0.0_f64, 1.0], [2.0, 3.0]]);
3492        let ds = EncodedDataset {
3493            headers: vec!["alpha".to_string(), "beta".to_string()],
3494            values,
3495            schema: DataSchema { columns: vec![] },
3496            column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3497        };
3498        let map = ds.column_map();
3499        assert_eq!(map["alpha"], 0);
3500        assert_eq!(map["beta"], 1);
3501        assert_eq!(map.len(), 2);
3502    }
3503
3504    // ── shared_prefix ─────────────────────────────────────────────────────────
3505
3506    #[test]
3507    fn shared_prefix_identical_strings() {
3508        assert_eq!(shared_prefix("hello", "hello"), 5);
3509    }
3510
3511    #[test]
3512    fn shared_prefix_no_common_prefix() {
3513        assert_eq!(shared_prefix("abc", "xyz"), 0);
3514    }
3515
3516    #[test]
3517    fn shared_prefix_partial_match() {
3518        assert_eq!(shared_prefix("foobar", "foobaz"), 5);
3519    }
3520
3521    #[test]
3522    fn shared_prefix_one_empty() {
3523        assert_eq!(shared_prefix("", "hello"), 0);
3524        assert_eq!(shared_prefix("hello", ""), 0);
3525    }
3526
3527    #[test]
3528    fn shared_prefix_both_empty() {
3529        assert_eq!(shared_prefix("", ""), 0);
3530    }
3531
3532    #[test]
3533    fn shared_prefix_shorter_string_is_prefix() {
3534        assert_eq!(shared_prefix("foo", "foobar"), 3);
3535    }
3536
3537    // ── detect_format ─────────────────────────────────────────────────────────
3538
3539    #[test]
3540    fn detect_format_csv() {
3541        let path = std::path::Path::new("data.csv");
3542        assert_eq!(detect_format(path).unwrap(), DataFormat::Csv);
3543    }
3544
3545    #[test]
3546    fn detect_format_tsv() {
3547        assert_eq!(
3548            detect_format(std::path::Path::new("data.tsv")).unwrap(),
3549            DataFormat::Tsv
3550        );
3551        assert_eq!(
3552            detect_format(std::path::Path::new("data.txt")).unwrap(),
3553            DataFormat::Tsv
3554        );
3555        assert_eq!(
3556            detect_format(std::path::Path::new("data.tab")).unwrap(),
3557            DataFormat::Tsv
3558        );
3559    }
3560
3561    #[test]
3562    fn detect_format_parquet() {
3563        assert_eq!(
3564            detect_format(std::path::Path::new("data.parquet")).unwrap(),
3565            DataFormat::Parquet
3566        );
3567        assert_eq!(
3568            detect_format(std::path::Path::new("data.pq")).unwrap(),
3569            DataFormat::Parquet
3570        );
3571        assert_eq!(
3572            detect_format(std::path::Path::new("data.pqt")).unwrap(),
3573            DataFormat::Parquet
3574        );
3575    }
3576
3577    #[test]
3578    fn detect_format_uppercase_extension() {
3579        assert_eq!(
3580            detect_format(std::path::Path::new("data.CSV")).unwrap(),
3581            DataFormat::Csv
3582        );
3583    }
3584
3585    #[test]
3586    fn detect_format_unknown_extension_is_error() {
3587        let err = detect_format(std::path::Path::new("data.json")).unwrap_err();
3588        let msg = format!("{err:?}");
3589        assert!(
3590            msg.contains("json") || msg.contains("unsupported"),
3591            "error should mention extension, got: {msg}"
3592        );
3593    }
3594
3595    // ── strip_categorical_sentinel ────────────────────────────────────────────
3596
3597    #[test]
3598    fn strip_categorical_sentinel_marked_cell() {
3599        // Sentinel is a single NUL character prefix
3600        let marked = "\u{0}hello";
3601        let (text, found) = strip_categorical_sentinel(marked);
3602        assert!(found);
3603        assert_eq!(text, "hello");
3604    }
3605
3606    #[test]
3607    fn strip_categorical_sentinel_unmarked_cell() {
3608        let (text, found) = strip_categorical_sentinel("plain");
3609        assert!(!found);
3610        assert_eq!(text, "plain");
3611    }
3612
3613    #[test]
3614    fn strip_categorical_sentinel_empty_string() {
3615        let (text, found) = strip_categorical_sentinel("");
3616        assert!(!found);
3617        assert_eq!(text, "");
3618    }
3619
3620    #[test]
3621    fn strip_categorical_sentinel_only_sentinel() {
3622        let s = "\u{0}";
3623        let (text, found) = strip_categorical_sentinel(s);
3624        assert!(found);
3625        assert_eq!(text, "");
3626    }
3627
3628    // ── projected_headers ─────────────────────────────────────────────────────
3629
3630    #[test]
3631    fn projected_headers_selects_by_index() {
3632        let all = vec![
3633            "a".to_string(),
3634            "b".to_string(),
3635            "c".to_string(),
3636            "d".to_string(),
3637        ];
3638        let selected = projected_headers(&all, &[1, 3]);
3639        assert_eq!(selected, vec!["b".to_string(), "d".to_string()]);
3640    }
3641
3642    #[test]
3643    fn projected_headers_empty_selection() {
3644        let all = vec!["x".to_string(), "y".to_string()];
3645        let selected = projected_headers(&all, &[]);
3646        assert!(selected.is_empty());
3647    }
3648
3649    #[test]
3650    fn projected_headers_all_indices() {
3651        let all = vec!["p".to_string(), "q".to_string()];
3652        let selected = projected_headers(&all, &[0, 1]);
3653        assert_eq!(selected, all);
3654    }
3655
3656    #[test]
3657    fn canonical_level_bits_collapses_signed_zero() {
3658        // The whole point of the helper: +0.0 and -0.0 name the same real
3659        // number (IEEE-754: 0.0 == -0.0) and MUST map to the same key, even
3660        // though their raw bit patterns differ (#2145 / #2146).
3661        let pos = 0.0_f64;
3662        let neg = -0.0_f64;
3663        assert_ne!(
3664            pos.to_bits(),
3665            neg.to_bits(),
3666            "precondition: raw bits differ"
3667        );
3668        assert_eq!(pos, neg, "precondition: numerically equal");
3669        assert_eq!(canonical_level_bits(pos), canonical_level_bits(neg));
3670        assert_eq!(canonical_level_bits(neg), 0.0_f64.to_bits());
3671        // -0.0 reached via ordinary arithmetic is handled the same way.
3672        assert_eq!(canonical_level_bits(-1.0 * 0.0), 0.0_f64.to_bits());
3673        assert_eq!(canonical_level_bits(0.0 - 0.0), 0.0_f64.to_bits());
3674    }
3675
3676    #[test]
3677    fn canonical_level_bits_is_bit_stable_on_ordinary_values() {
3678        // Every ordinary finite value keeps its raw key — the helper must not
3679        // perturb the identity of any genuine level.
3680        for &v in &[
3681            1.0_f64,
3682            -1.0,
3683            2.5,
3684            -3.75,
3685            1e300,
3686            -1e-300,
3687            f64::MIN,
3688            f64::MAX,
3689        ] {
3690            assert_eq!(canonical_level_bits(v), v.to_bits(), "value {v}");
3691        }
3692        // Distinct real values keep distinct keys.
3693        assert_ne!(canonical_level_bits(1.0), canonical_level_bits(2.0));
3694        assert_ne!(canonical_level_bits(0.0), canonical_level_bits(1.0));
3695        // Signed infinities are distinct (they are distinct real limits).
3696        assert_ne!(
3697            canonical_level_bits(f64::INFINITY),
3698            canonical_level_bits(f64::NEG_INFINITY)
3699        );
3700    }
3701
3702    #[test]
3703    fn canonical_level_bits_collapses_nan_payloads() {
3704        // Every NaN encoding denotes "not a number"; they collapse to one key.
3705        let a = f64::NAN;
3706        let b = f64::from_bits(0x7ff8_0000_0000_0001); // a different NaN payload
3707        let c = -f64::NAN; // sign-bit-set NaN
3708        assert!(a.is_nan() && b.is_nan() && c.is_nan());
3709        assert_eq!(canonical_level_bits(a), canonical_level_bits(b));
3710        assert_eq!(canonical_level_bits(a), canonical_level_bits(c));
3711    }
3712
3713    #[test]
3714    fn canonical_level_bits_is_idempotent() {
3715        // Re-canonicalizing an already-canonical key is a no-op — the property
3716        // the frozen-level resolution paths rely on.
3717        for &v in &[0.0_f64, -0.0, 1.0, -2.0, f64::NAN] {
3718            let once = canonical_level_bits(v);
3719            let twice = canonical_level_bits(f64::from_bits(once));
3720            assert_eq!(once, twice, "value {v}");
3721        }
3722    }
3723}