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
53pub fn encode_optional_categorical_column(
63 name: &str,
64 column: &[Option<&str>],
65) -> Result<(SchemaColumn, Vec<f64>), DataError> {
66 if column.is_empty() {
67 return Err(DataError::EmptyInput {
68 reason: "table data cannot be empty".to_string(),
69 });
70 }
71
72 let mut levels = Vec::new();
73 for (row, label) in column.iter().enumerate() {
74 let Some(label) = label else {
75 continue;
76 };
77 let label = label.trim();
78 if label.is_empty() {
79 return Err(DataError::EmptyInput {
80 reason: format!("empty field at row {}, column '{name}'", row + 1),
81 });
82 }
83 levels.push(label.to_string());
84 }
85 sort_levels_canonical(&mut levels);
86 levels.dedup();
87 let level_map = levels
88 .iter()
89 .enumerate()
90 .map(|(index, level)| (level.as_str(), index as f64))
91 .collect::<HashMap<_, _>>();
92 let values = column
93 .iter()
94 .map(|value| match value {
95 None => Ok(f64::NAN),
96 Some(label) => level_map.get(label.trim()).copied().ok_or_else(|| {
97 DataError::EncodingFailure {
98 reason: format!(
99 "internal: level '{}' missing from freshly built map for column '{name}'",
100 label.trim()
101 ),
102 }
103 }),
104 })
105 .collect::<Result<Vec<_>, _>>()?;
106
107 Ok((
108 SchemaColumn {
109 name: name.to_string(),
110 kind: ColumnKindTag::Categorical,
111 levels,
112 },
113 values,
114 ))
115}
116
117#[inline]
147pub fn canonical_level_bits(v: f64) -> u64 {
148 if v == 0.0 {
149 0.0_f64.to_bits()
151 } else if v.is_nan() {
152 f64::NAN.to_bits()
154 } else {
155 v.to_bits()
156 }
157}
158
159#[derive(Debug, Clone)]
170pub enum DataError {
171 SchemaMismatch { reason: String },
176 ParseError { reason: String },
180 EncodingFailure { reason: String },
184 EmptyInput { reason: String },
187 InvalidValue { reason: String },
190 DegenerateColumn { column: String, problem: String },
194 ColumnNotFound {
201 name: String,
203 role: Option<String>,
207 available: Vec<String>,
209 similar: Vec<String>,
212 tsv_hint: bool,
217 },
218}
219
220impl DataError {
221 #[must_use]
228 fn with_source_path(self, path: &Path) -> Self {
229 let qualify = |reason: String| {
230 if reason.contains(&path.display().to_string()) {
231 reason
232 } else {
233 format!("data file '{}': {reason}", path.display())
234 }
235 };
236 match self {
237 Self::SchemaMismatch { reason } => Self::SchemaMismatch { reason: qualify(reason) },
238 Self::ParseError { reason } => Self::ParseError { reason: qualify(reason) },
239 Self::EncodingFailure { reason } => Self::EncodingFailure { reason: qualify(reason) },
240 Self::EmptyInput { reason } => Self::EmptyInput { reason: qualify(reason) },
241 Self::InvalidValue { reason } => Self::InvalidValue { reason: qualify(reason) },
242 column @ Self::ColumnNotFound { .. } => column,
243 degenerate @ Self::DegenerateColumn { .. } => degenerate,
244 }
245 }
246
247 #[must_use]
250 pub fn advice(&self) -> Option<String> {
251 match self {
252 Self::SchemaMismatch { .. } => Some(
253 "Verify the new data has the same columns and types as the training data \
254 and that the formula terms match."
255 .to_string(),
256 ),
257 Self::ParseError { .. }
258 | Self::EncodingFailure { .. }
259 | Self::EmptyInput { .. }
260 | Self::InvalidValue { .. }
261 | Self::DegenerateColumn { .. }
262 | Self::ColumnNotFound { .. } => None,
263 }
264 }
265
266 pub fn column_not_found(
272 col_map: &HashMap<String, usize>,
273 name: &str,
274 role: Option<&str>,
275 ) -> Self {
276 let target_lower = name.to_lowercase();
277 let mut similar: Vec<String> = col_map
278 .keys()
279 .filter(|k| {
280 let k_lower = k.to_lowercase();
281 k_lower.contains(&target_lower)
282 || target_lower.contains(&k_lower)
283 || shared_prefix(&k_lower, &target_lower) >= 3
284 })
285 .cloned()
286 .collect();
287 similar.sort_unstable();
288 let mut available: Vec<String> = col_map.keys().cloned().collect();
289 available.sort_unstable();
290 let tsv_hint = available.len() == 1 && available[0].contains('\t');
291 Self::ColumnNotFound {
292 name: name.to_string(),
293 role: role.map(str::to_string),
294 available,
295 similar,
296 tsv_hint,
297 }
298 }
299}
300
301impl fmt::Display for DataError {
302 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
303 match self {
304 DataError::SchemaMismatch { reason }
305 | DataError::ParseError { reason }
306 | DataError::EncodingFailure { reason }
307 | DataError::EmptyInput { reason }
308 | DataError::InvalidValue { reason } => f.write_str(reason),
309 DataError::DegenerateColumn { column, problem } => {
310 write!(f, "column '{column}' {problem}")
311 }
312 DataError::ColumnNotFound {
313 name,
314 role,
315 available,
316 similar,
317 tsv_hint,
318 } => {
319 let label = match role {
320 Some(r) => format!("{r} column '{name}'"),
321 None => format!("column '{name}'"),
322 };
323 let tsv_suffix = if *tsv_hint {
324 " — your file appears to be tab-separated; gam expects comma-separated CSV. \
325 Replace tabs with commas, or pre-convert with `tr '\\t' ',' < file.tsv > file.csv`."
326 } else {
327 ""
328 };
329 if similar.is_empty() {
330 write!(
331 f,
332 "{label} not found in data. Available columns: [{}]{tsv_suffix}",
333 available.join(", ")
334 )
335 } else {
336 write!(
337 f,
338 "{label} not found in data. Did you mean one of [{}]? Full list: [{}]{tsv_suffix}",
339 similar.join(", "),
340 available.join(", ")
341 )
342 }
343 }
344 }
345 }
346}
347
348impl std::error::Error for DataError {}
349
350impl From<DataError> for String {
351 fn from(err: DataError) -> String {
352 err.to_string()
353 }
354}
355
356#[derive(Clone, Debug, Serialize, Deserialize)]
361pub struct DataSchema {
362 pub columns: Vec<SchemaColumn>,
363}
364
365#[derive(Clone, Debug, Serialize, Deserialize)]
366pub struct SchemaColumn {
367 pub name: String,
368 pub kind: ColumnKindTag,
369 #[serde(default)]
370 pub levels: Vec<String>,
371}
372
373#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
374#[serde(rename_all = "kebab-case")]
375pub enum ColumnKindTag {
376 Continuous,
377 Binary,
378 Categorical,
379}
380
381#[derive(Clone, Debug, Eq, PartialEq)]
382pub enum UnseenCategoryPolicy {
383 Error,
384 EncodeUnknownForColumns(HashSet<String>),
385}
386
387impl UnseenCategoryPolicy {
388 pub fn encode_unknown_for_columns(columns: HashSet<String>) -> Self {
389 if columns.is_empty() {
390 Self::Error
391 } else {
392 Self::EncodeUnknownForColumns(columns)
393 }
394 }
395
396 fn unseen_code_for(&self, column_name: &str, level_count: usize) -> Option<f64> {
397 match self {
398 Self::Error => None,
399 Self::EncodeUnknownForColumns(columns) => {
400 columns.contains(column_name).then_some(level_count as f64)
401 }
402 }
403 }
404}
405
406#[derive(Clone, Debug)]
407pub struct EncodedDataset {
408 pub headers: Vec<String>,
409 pub values: Array2<f64>,
410 pub schema: DataSchema,
411 pub column_kinds: Vec<ColumnKindTag>,
412}
413
414impl EncodedDataset {
415 pub fn validate_fit_boundary(&self) -> Result<(), DataError> {
427 if self.headers.is_empty() {
428 return Err(DataError::DegenerateColumn {
429 column: "<table>".to_string(),
430 problem: "has no columns".to_string(),
431 });
432 }
433 let mut seen = HashSet::with_capacity(self.headers.len());
434 for name in &self.headers {
435 if !seen.insert(name.as_str()) {
436 return Err(DataError::DegenerateColumn {
437 column: name.clone(),
438 problem: "has a duplicate name".to_string(),
439 });
440 }
441 }
442 if self.values.nrows() == 0 {
443 return Err(DataError::DegenerateColumn {
444 column: "<table>".to_string(),
445 problem: "has no observations".to_string(),
446 });
447 }
448 if self.values.ncols() != self.headers.len() {
449 return Err(DataError::SchemaMismatch {
450 reason: format!(
451 "table has {} headers but {} value columns",
452 self.headers.len(),
453 self.values.ncols()
454 ),
455 });
456 }
457 for (index, name) in self.headers.iter().enumerate() {
458 if self.column_kinds.get(index) == Some(&ColumnKindTag::Categorical)
459 && self
460 .schema
461 .columns
462 .get(index)
463 .is_some_and(|column| column.levels.len() < 2)
464 {
465 return Err(DataError::DegenerateColumn {
466 column: name.clone(),
467 problem: "is a factor with fewer than two levels".to_string(),
468 });
469 }
470 let column = self.values.column(index);
471 let finite_count = column.iter().filter(|value| value.is_finite()).count();
472 if finite_count == 1 && column.len() > 1 {
473 return Err(DataError::DegenerateColumn {
474 column: name.clone(),
475 problem: "has only one non-missing value".to_string(),
476 });
477 }
478 if let Some((row, value)) = column
479 .iter()
480 .enumerate()
481 .find(|(_, value)| !value.is_finite())
482 {
483 return Err(DataError::DegenerateColumn {
484 column: name.clone(),
485 problem: format!("has non-finite value {value} at row {}", row + 1),
486 });
487 }
488 }
489 Ok(())
490 }
491
492 pub fn column_map(&self) -> HashMap<String, usize> {
493 self.headers
494 .iter()
495 .enumerate()
496 .map(|(index, header)| (header.clone(), index))
497 .collect()
498 }
499
500 pub fn feature_ranges(&self) -> Vec<(f64, f64)> {
506 self.values
513 .axis_iter(Axis(1))
514 .into_par_iter()
515 .map(|col| {
516 let (lo, hi) =
517 col.iter()
518 .fold((f64::INFINITY, f64::NEG_INFINITY), |(lo, hi), &v| {
519 if v.is_finite() {
520 (lo.min(v), hi.max(v))
521 } else {
522 (lo, hi)
523 }
524 });
525 if !lo.is_finite() || !hi.is_finite() {
526 (0.0, 0.0)
527 } else {
528 (lo, hi)
529 }
530 })
531 .collect()
532 }
533}
534
535fn shared_prefix(a: &str, b: &str) -> usize {
536 a.chars()
537 .zip(b.chars())
538 .take_while(|(ca, cb)| ca == cb)
539 .count()
540}
541
542#[derive(Clone, Copy, Debug, Eq, PartialEq)]
547enum DataFormat {
548 Csv,
549 Tsv,
550 Parquet,
551}
552
553fn detect_format(path: &Path) -> Result<DataFormat, DataError> {
554 let ext = path
555 .extension()
556 .and_then(|s| s.to_str())
557 .unwrap_or_default()
558 .to_ascii_lowercase();
559 match ext.as_str() {
560 "csv" => Ok(DataFormat::Csv),
561 "tsv" | "txt" | "tab" => Ok(DataFormat::Tsv),
562 "parquet" | "pq" | "pqt" => Ok(DataFormat::Parquet),
563 other => Err(DataError::ParseError {
564 reason: format!(
565 "unsupported data file extension '.{other}'; expected csv, tsv, txt, parquet, or pq: '{}'",
566 path.display()
567 ),
568 }),
569 }
570}
571
572pub fn load_dataset_projected(
577 path: &Path,
578 requested_columns: &[String],
579) -> Result<EncodedDataset, DataError> {
580 load_dataset_projected_with_categorical_roles(path, requested_columns, &HashSet::new())
581}
582
583pub fn load_dataset_projected_with_categorical_roles(
605 path: &Path,
606 requested_columns: &[String],
607 categorical_roles: &HashSet<&str>,
608) -> Result<EncodedDataset, DataError> {
609 (match detect_format(path)? {
610 DataFormat::Csv => {
611 load_delimited_inferred(path, b',', requested_columns, categorical_roles)
612 }
613 DataFormat::Tsv => {
614 load_delimited_inferred(path, b'\t', requested_columns, categorical_roles)
615 }
616 DataFormat::Parquet => load_parquet_inferred(path, requested_columns, categorical_roles),
617 })
618 .map_err(|error| error.with_source_path(path))
619}
620
621pub fn load_datasetwith_schema_projected(
622 path: &Path,
623 schema: &DataSchema,
624 unseen_policy: UnseenCategoryPolicy,
625 requested_columns: &[String],
626) -> Result<EncodedDataset, DataError> {
627 (match detect_format(path)? {
628 DataFormat::Csv => {
629 load_delimited_with_schema(path, b',', schema, unseen_policy, requested_columns)
630 }
631 DataFormat::Tsv => {
632 load_delimited_with_schema(path, b'\t', schema, unseen_policy, requested_columns)
633 }
634 DataFormat::Parquet => {
635 load_parquet_with_schema(path, schema, unseen_policy, requested_columns)
636 }
637 })
638 .map_err(|error| error.with_source_path(path))
639}
640
641pub fn load_csvwith_inferred_schema(path: &Path) -> Result<EncodedDataset, DataError> {
646 load_delimited_inferred(path, b',', &[], &HashSet::new())
647 .map_err(|error| error.with_source_path(path))
648}
649
650pub const CATEGORICAL_CELL_SENTINEL: char = '\u{0}';
665
666pub fn strip_categorical_sentinel(cell: &str) -> (&str, bool) {
669 match cell.strip_prefix(CATEGORICAL_CELL_SENTINEL) {
670 Some(rest) => (rest, true),
671 None => (cell, false),
672 }
673}
674
675fn resolve_requested_columns(
676 all_headers: &[String],
677 requested_columns: &[String],
678) -> Result<Vec<usize>, DataError> {
679 if requested_columns.is_empty() {
680 return Ok((0..all_headers.len()).collect());
681 }
682
683 let requested_set: HashSet<&str> = requested_columns.iter().map(String::as_str).collect();
684 let mut selected = Vec::with_capacity(requested_set.len());
685 for (idx, name) in all_headers.iter().enumerate() {
686 if requested_set.contains(name.as_str()) {
687 selected.push(idx);
688 }
689 }
690
691 if selected.len() != requested_set.len() {
692 let available_map: HashMap<String, usize> = all_headers
693 .iter()
694 .enumerate()
695 .map(|(index, header)| (header.clone(), index))
696 .collect();
697 let missing = requested_columns
698 .iter()
699 .filter(|name| !available_map.contains_key(name.as_str()))
700 .map(|name| {
701 DataError::column_not_found(&available_map, name, Some("requested")).to_string()
702 })
703 .collect::<Vec<_>>();
704 return Err(DataError::SchemaMismatch {
705 reason: missing.join("; "),
706 });
707 }
708
709 Ok(selected)
710}
711
712fn projected_headers(all_headers: &[String], selected_indices: &[usize]) -> Vec<String> {
713 selected_indices
714 .iter()
715 .map(|&idx| all_headers[idx].clone())
716 .collect()
717}
718
719fn load_delimited_inferred(
720 path: &Path,
721 delimiter: u8,
722 requested_columns: &[String],
723 categorical_roles: &HashSet<&str>,
724) -> Result<EncodedDataset, DataError> {
725 let t_open = std::time::Instant::now();
726 let mut rdr = ReaderBuilder::new()
727 .has_headers(true)
728 .delimiter(delimiter)
729 .from_path(path)
730 .map_err(|e| DataError::ParseError {
731 reason: format!("failed to open '{}': {e}", path.display()),
732 })?;
733
734 let all_headers: Vec<String> = rdr
735 .headers()
736 .map_err(|e| DataError::ParseError {
737 reason: format!("failed to read headers: {e}"),
738 })?
739 .iter()
740 .map(|s| s.trim().to_string())
741 .collect();
742 if all_headers.is_empty() {
743 return Err(DataError::EmptyInput {
744 reason: "file has no headers".to_string(),
745 });
746 }
747 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
748 let headers = projected_headers(&all_headers, &selected_indices);
749 let p = headers.len();
750 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
751 if open_ms > 100.0 {
752 log::info!(
753 "[DATA-LOAD] delim_open+headers | n_headers={} | n_proj={} | {:.1}ms",
754 all_headers.len(),
755 p,
756 open_ms
757 );
758 }
759
760 let mut inference = vec![DelimitedInferenceState::default(); p];
764 let mut total_rows: usize = 0;
765 let t_stream = std::time::Instant::now();
766 let mut record = StringRecord::new();
767 while rdr
768 .read_record(&mut record)
769 .map_err(|e| DataError::ParseError {
770 reason: format!("failed reading row: {e}"),
771 })?
772 {
773 if record.len() != all_headers.len() {
774 return Err(DataError::SchemaMismatch {
775 reason: format!(
776 "row width mismatch at row {}: got {} fields, expected {}",
777 total_rows + 1,
778 record.len(),
779 all_headers.len()
780 ),
781 });
782 }
783 total_rows += 1;
784 for (j, &selected_idx) in selected_indices.iter().enumerate() {
785 inference[j].observe(
786 record
787 .get(selected_idx)
788 .expect("record width was checked against the header row above")
789 .trim(),
790 total_rows,
791 &headers[j],
792 )?;
793 }
794 }
795
796 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
797 if stream_ms > 100.0 {
798 log::info!(
799 "[DATA-LOAD] delim_stream | n_rows={} | n_cols={} | {:.1}ms",
800 total_rows,
801 p,
802 stream_ms
803 );
804 }
805
806 if total_rows == 0 {
807 return Err(DataError::EmptyInput {
808 reason: "file has no rows".to_string(),
809 });
810 }
811
812 let t_schema = std::time::Instant::now();
813 let column_kinds = inference
814 .iter()
815 .enumerate()
816 .map(|(j, state)| state.kind(categorical_roles.contains(headers[j].as_str())))
817 .collect::<Vec<_>>();
818 let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
819 if schema_ms > 100.0 {
820 let n_cat = column_kinds
821 .iter()
822 .filter(|k| matches!(k, ColumnKindTag::Categorical))
823 .count();
824 log::info!(
825 "[DATA-LOAD] delim_convert+infer | n_cols={} | n_cat={} | {:.1}ms",
826 p,
827 n_cat,
828 schema_ms
829 );
830 }
831
832 let t_assemble = std::time::Instant::now();
837 let mut values = Array2::<f64>::zeros((total_rows, p));
838 let mut categorical_encoders = (0..p)
839 .map(|j| {
840 matches!(column_kinds[j], ColumnKindTag::Categorical).then(CategoricalEncoder::default)
841 })
842 .collect::<Vec<_>>();
843 let mut encode_rdr = ReaderBuilder::new()
844 .has_headers(true)
845 .delimiter(delimiter)
846 .from_path(path)
847 .map_err(|e| DataError::ParseError {
848 reason: format!("failed to reopen '{}': {e}", path.display()),
849 })?;
850 encode_rdr.headers().map_err(|e| DataError::ParseError {
851 reason: format!("failed to reread headers: {e}"),
852 })?;
853 let mut encoded_rows = 0usize;
854 while encode_rdr
855 .read_record(&mut record)
856 .map_err(|e| DataError::ParseError {
857 reason: format!("failed reading row: {e}"),
858 })?
859 {
860 if record.len() != all_headers.len() {
861 return Err(DataError::SchemaMismatch {
862 reason: format!(
863 "row width mismatch at row {}: got {} fields, expected {}",
864 encoded_rows + 1,
865 record.len(),
866 all_headers.len()
867 ),
868 });
869 }
870 if encoded_rows >= total_rows {
871 return Err(DataError::SchemaMismatch {
872 reason: "data file changed while its schema was being discovered".to_string(),
873 });
874 }
875 for (j, &selected_idx) in selected_indices.iter().enumerate() {
876 let raw = record
877 .get(selected_idx)
878 .expect("record width was checked against the header row above")
879 .trim();
880 values[[encoded_rows, j]] = match column_kinds[j] {
881 ColumnKindTag::Continuous | ColumnKindTag::Binary => {
882 parse_inferred_numeric_cell(raw, encoded_rows + 1, &headers[j])?
883 }
884 ColumnKindTag::Categorical => {
885 if raw.is_empty() {
886 return Err(DataError::EmptyInput {
887 reason: format!(
888 "empty field at row {}, column '{}'",
889 encoded_rows + 1,
890 &headers[j]
891 ),
892 });
893 }
894 categorical_encoders[j]
895 .as_mut()
896 .expect("categorical encoder")
897 .encode(raw) as f64
898 }
899 };
900 }
901 encoded_rows += 1;
902 }
903 if encoded_rows != total_rows {
904 return Err(DataError::SchemaMismatch {
905 reason: "data file changed while its schema was being discovered".to_string(),
906 });
907 }
908
909 let mut levels = vec![Vec::<String>::new(); p];
910 for (j, encoder) in categorical_encoders.into_iter().enumerate() {
911 if let Some(encoder) = encoder {
912 levels[j] = encoder.finish(values.column_mut(j), LevelOrder::Canonical);
913 }
914 }
915 let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
916 if assemble_ms > 100.0 {
917 log::info!(
918 "[DATA-LOAD] delim_assemble_array2 | n_rows={} | n_cols={} | {:.1}ms",
919 total_rows,
920 p,
921 assemble_ms
922 );
923 }
924
925 let schema = DataSchema {
926 columns: headers
927 .iter()
928 .enumerate()
929 .map(|(j, name)| SchemaColumn {
930 name: name.clone(),
931 kind: column_kinds[j],
932 levels: std::mem::take(&mut levels[j]),
933 })
934 .collect(),
935 };
936 Ok(EncodedDataset {
937 headers,
938 values,
939 schema,
940 column_kinds,
941 })
942}
943
944#[derive(Clone, Copy)]
945struct DelimitedInferenceState {
946 all_numeric: bool,
947 all_binary: bool,
948 saw_numeric: bool,
952}
953
954impl Default for DelimitedInferenceState {
955 fn default() -> Self {
956 Self {
957 all_numeric: true,
958 all_binary: true,
959 saw_numeric: false,
960 }
961 }
962}
963
964impl DelimitedInferenceState {
965 fn observe(&mut self, raw: &str, row: usize, header: &str) -> Result<(), DataError> {
966 if raw.is_empty() {
967 return Err(DataError::EmptyInput {
968 reason: format!("empty field at row {row}, column '{header}'"),
969 });
970 }
971 if is_missing_marker(raw) {
975 return Ok(());
976 }
977 match raw.parse::<f64>() {
978 Ok(value) => {
979 self.saw_numeric = true;
980 if !value.is_finite() {
981 return Err(DataError::InvalidValue {
982 reason: format!("non-finite value at row {row}, column '{header}'"),
983 });
984 }
985 if (value - 0.0).abs() >= 1e-12 && (value - 1.0).abs() >= 1e-12 {
986 self.all_binary = false;
987 }
988 }
989 Err(_) => {
990 self.all_numeric = false;
991 self.all_binary = false;
992 }
993 }
994 Ok(())
995 }
996
997 fn kind(self, force_categorical: bool) -> ColumnKindTag {
998 if force_categorical || !(self.all_numeric && self.saw_numeric) {
999 ColumnKindTag::Categorical
1000 } else if self.all_binary {
1001 ColumnKindTag::Binary
1002 } else {
1003 ColumnKindTag::Continuous
1004 }
1005 }
1006}
1007
1008fn is_missing_marker(raw: &str) -> bool {
1025 matches!(
1026 raw.trim().to_ascii_uppercase().as_str(),
1027 "NA" | "N/A" | "NULL"
1028 )
1029}
1030
1031fn parse_inferred_numeric_cell(raw: &str, row: usize, header: &str) -> Result<f64, DataError> {
1032 if raw.is_empty() {
1033 return Err(DataError::EmptyInput {
1034 reason: format!("empty field at row {row}, column '{header}'"),
1035 });
1036 }
1037 if is_missing_marker(raw) {
1042 return Ok(f64::NAN);
1043 }
1044 let value = raw
1045 .parse::<f64>()
1046 .map_err(|error| DataError::EncodingFailure {
1047 reason: format!(
1048 "failed to parse numeric value '{raw}' at row {row}, column '{header}': {error}"
1049 ),
1050 })?;
1051 if !value.is_finite() {
1052 return Err(DataError::InvalidValue {
1053 reason: format!("non-finite value at row {row}, column '{header}'"),
1054 });
1055 }
1056 Ok(value)
1057}
1058
1059#[derive(Clone, Copy)]
1060enum LevelOrder {
1061 Encounter,
1062 Canonical,
1063}
1064
1065#[derive(Default)]
1074struct CategoricalEncoder {
1075 encounter_codes: HashMap<String, usize>,
1076}
1077
1078impl CategoricalEncoder {
1079 fn encode(&mut self, label: &str) -> usize {
1080 if let Some(&code) = self.encounter_codes.get(label) {
1081 return code;
1082 }
1083 let code = self.encounter_codes.len();
1084 self.encounter_codes.insert(label.to_owned(), code);
1085 code
1086 }
1087
1088 fn finish(self, mut encoded: ArrayViewMut1<'_, f64>, order: LevelOrder) -> Vec<String> {
1089 match order {
1090 LevelOrder::Encounter => {
1091 let mut levels = std::iter::repeat_with(|| None)
1092 .take(self.encounter_codes.len())
1093 .collect::<Vec<Option<String>>>();
1094 for (level, old_code) in self.encounter_codes {
1095 levels[old_code] = Some(level);
1096 }
1097 levels
1098 .into_iter()
1099 .map(|level| level.expect("encounter code must name one level"))
1100 .collect()
1101 }
1102 LevelOrder::Canonical => {
1103 let mut levels_with_old_codes =
1104 self.encounter_codes.into_iter().collect::<Vec<_>>();
1105 levels_with_old_codes
1106 .sort_by(|(a, _), (b, _)| natural_level_cmp(a.as_str(), b.as_str()));
1107 let mut remap = vec![0usize; levels_with_old_codes.len()];
1108 for (new_code, (_, old_code)) in levels_with_old_codes.iter().enumerate() {
1109 remap[*old_code] = new_code;
1110 }
1111 for code in encoded.iter_mut() {
1112 if code.is_finite() {
1113 *code = remap[*code as usize] as f64;
1114 }
1115 }
1116 levels_with_old_codes
1117 .into_iter()
1118 .map(|(level, _)| level)
1119 .collect()
1120 }
1121 }
1122 }
1123}
1124
1125fn load_delimited_with_schema(
1126 path: &Path,
1127 delimiter: u8,
1128 schema: &DataSchema,
1129 unseen_policy: UnseenCategoryPolicy,
1130 requested_columns: &[String],
1131) -> Result<EncodedDataset, DataError> {
1132 let t_open = std::time::Instant::now();
1133 let mut rdr = ReaderBuilder::new()
1134 .has_headers(true)
1135 .delimiter(delimiter)
1136 .from_path(path)
1137 .map_err(|e| DataError::ParseError {
1138 reason: format!("failed to open '{}': {e}", path.display()),
1139 })?;
1140
1141 let all_headers: Vec<String> = rdr
1142 .headers()
1143 .map_err(|e| DataError::ParseError {
1144 reason: format!("failed to read headers: {e}"),
1145 })?
1146 .iter()
1147 .map(|s| s.trim().to_string())
1148 .collect();
1149 if all_headers.is_empty() {
1150 return Err(DataError::EmptyInput {
1151 reason: "file has no headers".to_string(),
1152 });
1153 }
1154 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
1155 let headers = projected_headers(&all_headers, &selected_indices);
1156 let p = headers.len();
1157 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
1158 if open_ms > 100.0 {
1159 log::info!(
1160 "[DATA-LOAD] delim_schema_open+headers | n_headers={} | n_proj={} | {:.1}ms",
1161 all_headers.len(),
1162 p,
1163 open_ms
1164 );
1165 }
1166
1167 let schema_byname: HashMap<&str, &SchemaColumn> = schema
1169 .columns
1170 .iter()
1171 .map(|c| (c.name.as_str(), c))
1172 .collect();
1173
1174 let mut col_meta = Vec::<ColMeta>::with_capacity(p);
1175 for name in &headers {
1176 if let Some(sc) = schema_byname.get(name.as_str()) {
1177 let level_map = if matches!(sc.kind, ColumnKindTag::Categorical) {
1178 Some(
1179 sc.levels
1180 .iter()
1181 .enumerate()
1182 .map(|(idx, v)| (v.as_str(), idx as f64))
1183 .collect::<HashMap<_, _>>(),
1184 )
1185 } else {
1186 None
1187 };
1188 col_meta.push(ColMeta {
1189 kind: sc.kind,
1190 level_map,
1191 schema_col: (*sc).clone(),
1192 });
1193 } else {
1194 col_meta.push(ColMeta {
1196 kind: ColumnKindTag::Continuous, level_map: None,
1198 schema_col: SchemaColumn {
1199 name: name.clone(),
1200 kind: ColumnKindTag::Continuous,
1201 levels: Vec::new(),
1202 },
1203 });
1204 }
1205 }
1206
1207 let needs_inference: Vec<bool> = headers
1209 .iter()
1210 .map(|h| !schema_byname.contains_key(h.as_str()))
1211 .collect();
1212
1213 if needs_inference.iter().all(|needs| !needs) {
1219 let t_stream = std::time::Instant::now();
1220 let mut flat_values = Vec::<f64>::new();
1221 let mut total_rows = 0usize;
1222 let mut record = StringRecord::new();
1223 while rdr
1224 .read_record(&mut record)
1225 .map_err(|e| DataError::ParseError {
1226 reason: format!("failed reading row: {e}"),
1227 })?
1228 {
1229 if record.len() != all_headers.len() {
1230 return Err(DataError::SchemaMismatch {
1231 reason: format!(
1232 "row width mismatch at row {}: got {} fields, expected {}",
1233 total_rows + 1,
1234 record.len(),
1235 all_headers.len()
1236 ),
1237 });
1238 }
1239 total_rows += 1;
1240 for j in 0..p {
1241 let raw = record
1242 .get(selected_indices[j])
1243 .expect("record width was checked against the header row above")
1244 .trim();
1245 flat_values.push(parse_cell_with_schema(
1246 raw,
1247 &col_meta[j],
1248 total_rows,
1249 &headers[j],
1250 &unseen_policy,
1251 )?);
1252 }
1253 }
1254 if total_rows == 0 {
1255 return Err(DataError::EmptyInput {
1256 reason: "file has no rows".to_string(),
1257 });
1258 }
1259 let values = Array2::from_shape_vec((total_rows, p), flat_values).map_err(|error| {
1260 DataError::EncodingFailure {
1261 reason: format!("failed to assemble schema-guided delimited matrix: {error}"),
1262 }
1263 })?;
1264 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
1265 if stream_ms > 100.0 {
1266 log::info!(
1267 "[DATA-LOAD] delim_schema_direct | n_rows={} | n_cols={} | {:.1}ms",
1268 total_rows,
1269 p,
1270 stream_ms
1271 );
1272 }
1273 let column_kinds = col_meta.iter().map(|meta| meta.kind).collect();
1274 let schema_out = DataSchema {
1275 columns: col_meta.into_iter().map(|meta| meta.schema_col).collect(),
1276 };
1277 return Ok(EncodedDataset {
1278 headers,
1279 values,
1280 schema: schema_out,
1281 column_kinds,
1282 });
1283 }
1284
1285 let mut inference = vec![DelimitedInferenceState::default(); p];
1288 let mut total_rows: usize = 0;
1289 let t_stream = std::time::Instant::now();
1290 let mut record = StringRecord::new();
1291 while rdr
1292 .read_record(&mut record)
1293 .map_err(|e| DataError::ParseError {
1294 reason: format!("failed reading row: {e}"),
1295 })?
1296 {
1297 if record.len() != all_headers.len() {
1298 return Err(DataError::SchemaMismatch {
1299 reason: format!(
1300 "row width mismatch at row {}: got {} fields, expected {}",
1301 total_rows + 1,
1302 record.len(),
1303 all_headers.len()
1304 ),
1305 });
1306 }
1307 total_rows += 1;
1308
1309 for j in 0..p {
1310 let raw = record
1311 .get(selected_indices[j])
1312 .expect("record width was checked against the header row above")
1313 .trim();
1314 if needs_inference[j] {
1315 inference[j].observe(raw, total_rows, &headers[j])?;
1316 } else {
1317 parse_cell_with_schema(raw, &col_meta[j], total_rows, &headers[j], &unseen_policy)?;
1318 }
1319 }
1320 }
1321
1322 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
1323 if stream_ms > 100.0 {
1324 let n_inf = needs_inference.iter().filter(|x| **x).count();
1325 log::info!(
1326 "[DATA-LOAD] delim_schema_stream | n_rows={} | n_cols={} | n_inf={} | {:.1}ms",
1327 total_rows,
1328 p,
1329 n_inf,
1330 stream_ms
1331 );
1332 }
1333
1334 if total_rows == 0 {
1335 return Err(DataError::EmptyInput {
1336 reason: "file has no rows".to_string(),
1337 });
1338 }
1339
1340 let t_finalize = std::time::Instant::now();
1341 for j in 0..p {
1342 if needs_inference[j] {
1343 let kind = inference[j].kind(false);
1344 col_meta[j].kind = kind;
1345 col_meta[j].schema_col.kind = kind;
1346 }
1347 }
1348 let finalize_ms = t_finalize.elapsed().as_secs_f64() * 1000.0;
1349 if finalize_ms > 100.0 {
1350 log::info!(
1351 "[DATA-LOAD] delim_schema_finalize | n_cols={} | {:.1}ms",
1352 p,
1353 finalize_ms
1354 );
1355 }
1356
1357 let t_assemble = std::time::Instant::now();
1361 let mut values = Array2::<f64>::zeros((total_rows, p));
1362 let mut inferred_encoders = (0..p)
1363 .map(|j| {
1364 (needs_inference[j] && matches!(col_meta[j].kind, ColumnKindTag::Categorical))
1365 .then(CategoricalEncoder::default)
1366 })
1367 .collect::<Vec<_>>();
1368 let mut encode_rdr = ReaderBuilder::new()
1369 .has_headers(true)
1370 .delimiter(delimiter)
1371 .from_path(path)
1372 .map_err(|e| DataError::ParseError {
1373 reason: format!("failed to reopen '{}': {e}", path.display()),
1374 })?;
1375 encode_rdr.headers().map_err(|e| DataError::ParseError {
1376 reason: format!("failed to reread headers: {e}"),
1377 })?;
1378 let mut encoded_rows = 0usize;
1379 while encode_rdr
1380 .read_record(&mut record)
1381 .map_err(|e| DataError::ParseError {
1382 reason: format!("failed reading row: {e}"),
1383 })?
1384 {
1385 if record.len() != all_headers.len() {
1386 return Err(DataError::SchemaMismatch {
1387 reason: format!(
1388 "row width mismatch at row {}: got {} fields, expected {}",
1389 encoded_rows + 1,
1390 record.len(),
1391 all_headers.len()
1392 ),
1393 });
1394 }
1395 if encoded_rows >= total_rows {
1396 return Err(DataError::SchemaMismatch {
1397 reason: "data file changed while its schema was being discovered".to_string(),
1398 });
1399 }
1400 for j in 0..p {
1401 let raw = record
1402 .get(selected_indices[j])
1403 .expect("record width was checked against the header row above")
1404 .trim();
1405 values[[encoded_rows, j]] = if !needs_inference[j] {
1406 parse_cell_with_schema(
1407 raw,
1408 &col_meta[j],
1409 encoded_rows + 1,
1410 &headers[j],
1411 &unseen_policy,
1412 )?
1413 } else {
1414 match col_meta[j].kind {
1415 ColumnKindTag::Continuous | ColumnKindTag::Binary => {
1416 parse_inferred_numeric_cell(raw, encoded_rows + 1, &headers[j])?
1417 }
1418 ColumnKindTag::Categorical => {
1419 if raw.is_empty() {
1420 return Err(DataError::EmptyInput {
1421 reason: format!(
1422 "empty field at row {}, column '{}'",
1423 encoded_rows + 1,
1424 &headers[j]
1425 ),
1426 });
1427 }
1428 let encoder = inferred_encoders[j]
1429 .as_mut()
1430 .expect("inferred categorical encoder");
1431 encoder.encode(raw) as f64
1432 }
1433 }
1434 };
1435 }
1436 encoded_rows += 1;
1437 }
1438 if encoded_rows != total_rows {
1439 return Err(DataError::SchemaMismatch {
1440 reason: "data file changed while its schema was being discovered".to_string(),
1441 });
1442 }
1443 for (j, encoder) in inferred_encoders.into_iter().enumerate() {
1444 if let Some(encoder) = encoder {
1445 col_meta[j].schema_col.levels =
1446 encoder.finish(values.column_mut(j), LevelOrder::Canonical);
1447 }
1448 }
1449 let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
1450 if assemble_ms > 100.0 {
1451 log::info!(
1452 "[DATA-LOAD] delim_schema_assemble | n_rows={} | n_cols={} | {:.1}ms",
1453 total_rows,
1454 p,
1455 assemble_ms
1456 );
1457 }
1458
1459 let column_kinds = col_meta.iter().map(|meta| meta.kind).collect();
1460 let schema_out = DataSchema {
1461 columns: col_meta.into_iter().map(|m| m.schema_col).collect(),
1462 };
1463 Ok(EncodedDataset {
1464 headers,
1465 values,
1466 schema: schema_out,
1467 column_kinds,
1468 })
1469}
1470
1471fn parse_cell_with_schema(
1472 raw: &str,
1473 meta: &ColMeta<'_>,
1474 row: usize,
1475 col_name: &str,
1476 unseen_policy: &UnseenCategoryPolicy,
1477) -> Result<f64, DataError> {
1478 let val = match meta.kind {
1479 ColumnKindTag::Continuous if is_missing_marker(raw) => f64::NAN,
1482 ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
1483 DataError::SchemaMismatch {
1484 reason: format!(
1485 "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
1486 col_name, row, raw, err
1487 ),
1488 }
1489 })?,
1490 ColumnKindTag::Binary if is_missing_marker(raw) => f64::NAN,
1491 ColumnKindTag::Binary => {
1492 let v = raw
1493 .parse::<f64>()
1494 .map_err(|err| DataError::SchemaMismatch {
1495 reason: format!(
1496 "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
1497 col_name, row, raw, err
1498 ),
1499 })?;
1500 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
1501 return Err(DataError::SchemaMismatch {
1502 reason: format!(
1503 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
1504 col_name, row, v
1505 ),
1506 });
1507 }
1508 v
1509 }
1510 ColumnKindTag::Categorical => {
1511 let map = meta
1512 .level_map
1513 .as_ref()
1514 .ok_or_else(|| DataError::EncodingFailure {
1515 reason: "internal categorical schema map missing".to_string(),
1516 })?;
1517 match map.get(raw) {
1518 Some(v) => *v,
1519 None => unseen_policy
1520 .unseen_code_for(col_name, meta.schema_col.levels.len())
1521 .ok_or_else(|| DataError::SchemaMismatch {
1522 reason: format!(
1523 "unseen level '{}' in categorical column '{}' at row {}",
1524 raw, col_name, row
1525 ),
1526 })?,
1527 }
1528 }
1529 };
1530 if !val.is_finite() && !is_missing_marker(raw) {
1534 return Err(DataError::InvalidValue {
1535 reason: format!("non-finite value at row {}, column '{}'", row, col_name),
1536 });
1537 }
1538 Ok(val)
1539}
1540
1541struct ColMeta<'a> {
1544 kind: ColumnKindTag,
1545 level_map: Option<HashMap<&'a str, f64>>,
1546 schema_col: SchemaColumn,
1547}
1548
1549fn arrow_field_is_string(dt: &arrow::datatypes::DataType) -> bool {
1562 use arrow::datatypes::DataType;
1563 match dt {
1564 DataType::Utf8 | DataType::LargeUtf8 => true,
1565 DataType::Dictionary(_, value_type) => arrow_field_is_string(value_type),
1566 _ => false,
1567 }
1568}
1569
1570fn write_arrow_numeric_values(
1571 values: impl IntoIterator<Item = Option<f64>>,
1572 mut output: ArrayViewMut1<'_, f64>,
1573 categorical_encoder: Option<&mut CategoricalEncoder>,
1574 all_binary: &mut bool,
1575 saw_numeric: &mut bool,
1576) {
1577 match categorical_encoder {
1578 Some(encoder) => {
1579 for (batch_row, value) in values.into_iter().enumerate() {
1580 output[batch_row] = match value.filter(|value| value.is_finite()) {
1581 Some(value) => encoder.encode(&value.to_string()) as f64,
1582 None => f64::NAN,
1583 };
1584 }
1585 }
1586 None => {
1587 for (batch_row, value) in values.into_iter().enumerate() {
1588 let Some(value) = value.filter(|value| value.is_finite()) else {
1589 output[batch_row] = f64::NAN;
1590 continue;
1591 };
1592 *saw_numeric = true;
1593 if (value - 0.0).abs() >= 1e-12 && (value - 1.0).abs() >= 1e-12 {
1594 *all_binary = false;
1595 }
1596 output[batch_row] = value;
1597 }
1598 }
1599 }
1600}
1601
1602fn arrow_dictionary_string_value_at<'a, K>(
1603 col: &'a dyn arrow::array::Array,
1604 index: usize,
1605 logical_row: usize,
1606 header: &str,
1607) -> Result<Option<&'a str>, DataError>
1608where
1609 K: arrow::datatypes::ArrowDictionaryKeyType,
1610{
1611 use arrow::array::DictionaryArray;
1612
1613 let dictionary = col
1614 .as_any()
1615 .downcast_ref::<DictionaryArray<K>>()
1616 .ok_or_else(|| DataError::EncodingFailure {
1617 reason: format!(
1618 "Arrow dictionary column '{}' did not match its declared key type",
1619 header
1620 ),
1621 })?;
1622 let Some(value_index) = dictionary.key(index) else {
1623 return Ok(None);
1624 };
1625 if value_index >= dictionary.values().len() {
1626 return Err(DataError::EncodingFailure {
1627 reason: format!(
1628 "Arrow dictionary column '{}' has out-of-range key {} at row {}",
1629 header, value_index, logical_row
1630 ),
1631 });
1632 }
1633 arrow_string_value_at(
1634 dictionary.values().as_ref(),
1635 value_index,
1636 logical_row,
1637 header,
1638 )
1639}
1640
1641fn arrow_string_value_at<'a>(
1644 col: &'a dyn arrow::array::Array,
1645 index: usize,
1646 logical_row: usize,
1647 header: &str,
1648) -> Result<Option<&'a str>, DataError> {
1649 use arrow::array::{LargeStringArray, StringArray};
1650 use arrow::datatypes::{
1651 DataType, Int8Type, Int16Type, Int32Type, Int64Type, UInt8Type, UInt16Type, UInt32Type,
1652 UInt64Type,
1653 };
1654
1655 if index >= col.len() {
1656 return Err(DataError::EncodingFailure {
1657 reason: format!(
1658 "Arrow string column '{}' has out-of-range index {} at row {}",
1659 header, index, logical_row
1660 ),
1661 });
1662 }
1663 if col.is_null(index) {
1664 return Ok(None);
1665 }
1666
1667 match col.data_type() {
1668 DataType::Utf8 => col
1669 .as_any()
1670 .downcast_ref::<StringArray>()
1671 .map(|array| Some(array.value(index)))
1672 .ok_or_else(|| DataError::EncodingFailure {
1673 reason: format!("Arrow column '{}' could not be read as Utf8", header),
1674 }),
1675 DataType::LargeUtf8 => col
1676 .as_any()
1677 .downcast_ref::<LargeStringArray>()
1678 .map(|array| Some(array.value(index)))
1679 .ok_or_else(|| DataError::EncodingFailure {
1680 reason: format!("Arrow column '{}' could not be read as LargeUtf8", header),
1681 }),
1682 DataType::Dictionary(key_type, _) => match key_type.as_ref() {
1683 DataType::Int8 => {
1684 arrow_dictionary_string_value_at::<Int8Type>(col, index, logical_row, header)
1685 }
1686 DataType::Int16 => {
1687 arrow_dictionary_string_value_at::<Int16Type>(col, index, logical_row, header)
1688 }
1689 DataType::Int32 => {
1690 arrow_dictionary_string_value_at::<Int32Type>(col, index, logical_row, header)
1691 }
1692 DataType::Int64 => {
1693 arrow_dictionary_string_value_at::<Int64Type>(col, index, logical_row, header)
1694 }
1695 DataType::UInt8 => {
1696 arrow_dictionary_string_value_at::<UInt8Type>(col, index, logical_row, header)
1697 }
1698 DataType::UInt16 => {
1699 arrow_dictionary_string_value_at::<UInt16Type>(col, index, logical_row, header)
1700 }
1701 DataType::UInt32 => {
1702 arrow_dictionary_string_value_at::<UInt32Type>(col, index, logical_row, header)
1703 }
1704 DataType::UInt64 => {
1705 arrow_dictionary_string_value_at::<UInt64Type>(col, index, logical_row, header)
1706 }
1707 other => Err(DataError::InvalidValue {
1708 reason: format!(
1709 "unsupported Arrow dictionary key type {:?} for column '{}'",
1710 other, header
1711 ),
1712 }),
1713 },
1714 other => Err(DataError::InvalidValue {
1715 reason: format!(
1716 "unsupported Arrow string column type {:?} for column '{}'",
1717 other, header
1718 ),
1719 }),
1720 }
1721}
1722
1723fn decode_arrow_batch_column_into(
1730 col: &dyn arrow::array::Array,
1731 base_row: usize,
1732 header: &str,
1733 is_string_col: bool,
1734 mut output: ArrayViewMut1<'_, f64>,
1735 mut categorical_encoder: Option<&mut CategoricalEncoder>,
1736 all_binary: &mut bool,
1737 saw_numeric: &mut bool,
1738) -> Result<(), DataError> {
1739 use arrow::array::{
1740 Array as _, BooleanArray, Float32Array, Float64Array, Int8Array, Int16Array, Int32Array,
1741 Int64Array, UInt8Array, UInt16Array, UInt32Array, UInt64Array,
1742 };
1743 use arrow::datatypes::DataType;
1744
1745 let n_rows = output.len();
1746 if col.len() != n_rows {
1747 return Err(DataError::SchemaMismatch {
1748 reason: format!(
1749 "Arrow column '{}' has {} rows, but its record batch has {}",
1750 header,
1751 col.len(),
1752 n_rows
1753 ),
1754 });
1755 }
1756 if is_string_col {
1757 let encoder =
1758 categorical_encoder
1759 .as_deref_mut()
1760 .ok_or_else(|| DataError::EncodingFailure {
1761 reason: format!("categorical Arrow encoder missing for column '{header}'"),
1762 })?;
1763 for batch_row in 0..n_rows {
1764 output[batch_row] = match arrow_string_value_at(
1765 col,
1766 batch_row,
1767 base_row + batch_row + 1,
1768 header,
1769 )? {
1770 Some("") | None => f64::NAN,
1777 Some(label) => encoder.encode(label) as f64,
1778 };
1779 }
1780 return Ok(());
1781 }
1782
1783 let decoded_col;
1790 let col: &dyn arrow::array::Array = if let DataType::Dictionary(_, value_type) = col.data_type()
1791 {
1792 decoded_col = arrow::compute::cast(col, value_type).map_err(|e| DataError::ParseError {
1793 reason: format!(
1794 "failed to decode dictionary-encoded numeric column '{}': {e}",
1795 header
1796 ),
1797 })?;
1798 decoded_col.as_ref()
1799 } else {
1800 col
1801 };
1802 macro_rules! write_primitive {
1803 ($array_type:ty, $convert:expr) => {{
1804 let array = col
1805 .as_any()
1806 .downcast_ref::<$array_type>()
1807 .expect("array type is the one this `col.data_type()` arm matched");
1808 write_arrow_numeric_values(
1809 (0..n_rows).map(|index| {
1810 (!array.is_null(index)).then(|| $convert(array.value(index)))
1811 }),
1812 output,
1813 categorical_encoder.as_deref_mut(),
1814 all_binary,
1815 saw_numeric,
1816 );
1817 Ok(())
1818 }};
1819 }
1820
1821 match col.data_type() {
1822 DataType::Float64 => write_primitive!(Float64Array, |value: f64| value),
1823 DataType::Float32 => write_primitive!(Float32Array, |value: f32| value as f64),
1824 DataType::Int64 => write_primitive!(Int64Array, |value: i64| value as f64),
1825 DataType::Int32 => write_primitive!(Int32Array, |value: i32| value as f64),
1826 DataType::Int16 => write_primitive!(Int16Array, |value: i16| value as f64),
1827 DataType::Int8 => write_primitive!(Int8Array, |value: i8| value as f64),
1828 DataType::UInt64 => write_primitive!(UInt64Array, |value: u64| value as f64),
1829 DataType::UInt32 => write_primitive!(UInt32Array, |value: u32| value as f64),
1830 DataType::UInt16 => write_primitive!(UInt16Array, |value: u16| value as f64),
1831 DataType::UInt8 => write_primitive!(UInt8Array, |value: u8| value as f64),
1832 DataType::Boolean => {
1833 let arr = col
1834 .as_any()
1835 .downcast_ref::<BooleanArray>()
1836 .expect("array type is BooleanArray in the DataType::Boolean arm");
1837 write_arrow_numeric_values(
1838 (0..n_rows).map(|index| {
1839 (!arr.is_null(index)).then(|| if arr.value(index) { 1.0 } else { 0.0 })
1840 }),
1841 output,
1842 categorical_encoder.as_deref_mut(),
1843 all_binary,
1844 saw_numeric,
1845 );
1846 Ok(())
1847 }
1848 other => Err(DataError::InvalidValue {
1849 reason: format!(
1850 "unsupported Arrow column type {:?} for column '{}'",
1851 other, header
1852 ),
1853 }),
1854 }
1855}
1856
1857pub fn encode_arrow_record_batch_reader_with_inferred_schema(
1871 reader: &mut dyn arrow::record_batch::RecordBatchReader,
1872 headers: Vec<String>,
1873) -> Result<EncodedDataset, DataError> {
1874 if headers.is_empty() {
1875 return Err(DataError::EmptyInput {
1876 reason: "Arrow table must have at least one header column".to_string(),
1877 });
1878 }
1879
1880 let mut seen_headers = HashSet::<&str>::with_capacity(headers.len());
1881 for (column, header) in headers.iter().enumerate() {
1882 if header.trim().is_empty() {
1883 return Err(DataError::EmptyInput {
1884 reason: format!("Arrow header at column {} cannot be empty", column + 1),
1885 });
1886 }
1887 if !seen_headers.insert(header.as_str()) {
1888 return Err(DataError::SchemaMismatch {
1889 reason: format!("duplicate Arrow header '{}'", header),
1890 });
1891 }
1892 }
1893
1894 let arrow_schema = reader.schema();
1895 let p = headers.len();
1896 if arrow_schema.fields().len() != p {
1897 return Err(DataError::SchemaMismatch {
1898 reason: format!(
1899 "Arrow schema has {} columns, but {} normalized headers were supplied",
1900 arrow_schema.fields().len(),
1901 p
1902 ),
1903 });
1904 }
1905
1906 let is_string_col = arrow_schema
1907 .fields()
1908 .iter()
1909 .map(|field| arrow_field_is_string(field.data_type()))
1910 .collect::<Vec<_>>();
1911 let mut all_binary = vec![true; p];
1912 let mut saw_numeric = vec![false; p];
1913 let mut categorical_encoders = is_string_col
1914 .iter()
1915 .map(|&is_string| is_string.then(CategoricalEncoder::default))
1916 .collect::<Vec<_>>();
1917 let mut encoded_values = Vec::<f64>::new();
1918 let mut rows_seen = 0usize;
1919
1920 for batch_result in reader {
1921 let batch = batch_result.map_err(|error| DataError::ParseError {
1922 reason: format!("failed to read Arrow record batch: {error}"),
1923 })?;
1924 if batch.num_columns() != p {
1925 return Err(DataError::SchemaMismatch {
1926 reason: format!(
1927 "Arrow record batch has {} columns, but {} normalized headers were supplied",
1928 batch.num_columns(),
1929 p
1930 ),
1931 });
1932 }
1933 for j in 0..p {
1934 let expected = arrow_schema.field(j).data_type();
1935 let actual = batch.column(j).data_type();
1936 if actual != expected {
1937 return Err(DataError::SchemaMismatch {
1938 reason: format!(
1939 "Arrow column '{}' changed type between schema and batch: expected {:?}, got {:?}",
1940 headers[j], expected, actual
1941 ),
1942 });
1943 }
1944 }
1945
1946 let n_rows = batch.num_rows();
1947 let batch_values = n_rows
1948 .checked_mul(p)
1949 .ok_or_else(|| DataError::EncodingFailure {
1950 reason: "Arrow batch dimensions do not fit in memory address space".to_string(),
1951 })?;
1952 let next_len = encoded_values
1953 .len()
1954 .checked_add(batch_values)
1955 .ok_or_else(|| DataError::EncodingFailure {
1956 reason: "Arrow dataset dimensions do not fit in memory address space".to_string(),
1957 })?;
1958 encoded_values
1959 .try_reserve(batch_values)
1960 .map_err(|error| DataError::EncodingFailure {
1961 reason: format!("failed to reserve Arrow dataset storage: {error}"),
1962 })?;
1963 let batch_offset = encoded_values.len();
1964 encoded_values.resize(next_len, 0.0);
1965 let mut batch_output = ndarray::ArrayViewMut2::from_shape(
1966 (n_rows, p),
1967 &mut encoded_values[batch_offset..next_len],
1968 )
1969 .map_err(|error| DataError::EncodingFailure {
1970 reason: format!("failed to shape Arrow batch output: {error}"),
1971 })?;
1972
1973 let decoded_columns = batch_output
1974 .axis_iter_mut(Axis(1))
1975 .into_par_iter()
1976 .zip(categorical_encoders.par_iter_mut())
1977 .zip(all_binary.par_iter_mut())
1978 .zip(saw_numeric.par_iter_mut())
1979 .enumerate()
1980 .map(|(j, (((output, encoder), column_all_binary), column_saw_numeric))| {
1981 decode_arrow_batch_column_into(
1982 batch.column(j).as_ref(),
1983 rows_seen,
1984 &headers[j],
1985 is_string_col[j],
1986 output,
1987 encoder.as_mut(),
1988 column_all_binary,
1989 column_saw_numeric,
1990 )
1991 })
1992 .collect::<Vec<_>>();
1993 for decoded in decoded_columns {
1994 decoded?;
1995 }
1996 rows_seen = rows_seen
1997 .checked_add(n_rows)
1998 .ok_or_else(|| DataError::EncodingFailure {
1999 reason: "Arrow row count does not fit in memory address space".to_string(),
2000 })?;
2001 }
2002
2003 if rows_seen == 0 {
2004 return Err(DataError::EmptyInput {
2005 reason: "Arrow table data cannot be empty".to_string(),
2006 });
2007 }
2008
2009 let mut values = Array2::from_shape_vec((rows_seen, p), encoded_values).map_err(|error| {
2010 DataError::EncodingFailure {
2011 reason: format!("failed to shape encoded Arrow dataset: {error}"),
2012 }
2013 })?;
2014 let mut levels = vec![Vec::<String>::new(); p];
2015 for (j, encoder) in categorical_encoders.into_iter().enumerate() {
2016 if let Some(encoder) = encoder {
2017 levels[j] = encoder.finish(values.column_mut(j), LevelOrder::Canonical);
2018 }
2019 }
2020
2021 let mut schema_columns = Vec::<SchemaColumn>::with_capacity(p);
2022 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2023 for (j, name) in headers.iter().enumerate() {
2024 let kind = if is_string_col[j] {
2025 ColumnKindTag::Categorical
2026 } else if all_binary[j] && saw_numeric[j] {
2027 ColumnKindTag::Binary
2028 } else {
2029 ColumnKindTag::Continuous
2030 };
2031 column_kinds.push(kind);
2032 schema_columns.push(SchemaColumn {
2033 name: name.clone(),
2034 kind,
2035 levels: std::mem::take(&mut levels[j]),
2036 });
2037 }
2038
2039 Ok(EncodedDataset {
2040 headers,
2041 values,
2042 schema: DataSchema {
2043 columns: schema_columns,
2044 },
2045 column_kinds,
2046 })
2047}
2048
2049fn load_parquet_inferred(
2050 path: &Path,
2051 requested_columns: &[String],
2052 categorical_roles: &HashSet<&str>,
2053) -> Result<EncodedDataset, DataError> {
2054 use parquet::arrow::{ProjectionMask, arrow_reader::ParquetRecordBatchReaderBuilder};
2055 use rayon::prelude::*;
2056 use std::fs::File;
2057
2058 let t_open = std::time::Instant::now();
2059 let file = File::open(path).map_err(|e| DataError::ParseError {
2060 reason: format!("failed to open parquet '{}': {e}", path.display()),
2061 })?;
2062 let builder =
2063 ParquetRecordBatchReaderBuilder::try_new(file).map_err(|e| DataError::ParseError {
2064 reason: format!("failed to read parquet metadata '{}': {e}", path.display()),
2065 })?;
2066
2067 let full_schema = builder.schema().clone();
2068 let all_headers: Vec<String> = full_schema
2069 .fields()
2070 .iter()
2071 .map(|f| f.name().clone())
2072 .collect();
2073 if all_headers.is_empty() {
2074 return Err(DataError::EmptyInput {
2075 reason: "parquet file has no columns".to_string(),
2076 });
2077 }
2078 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
2079 let headers = projected_headers(&all_headers, &selected_indices);
2080 let selected_fields = selected_indices
2081 .iter()
2082 .map(|&idx| full_schema.fields()[idx].clone())
2083 .collect::<Vec<_>>();
2084 let total_rows =
2085 usize::try_from(builder.metadata().file_metadata().num_rows()).map_err(|_| {
2086 DataError::ParseError {
2087 reason: "parquet row count does not fit in memory address space".to_string(),
2088 }
2089 })?;
2090 if total_rows == 0 {
2091 return Err(DataError::EmptyInput {
2092 reason: "parquet file has no rows".to_string(),
2093 });
2094 }
2095 let projection =
2096 ProjectionMask::roots(builder.parquet_schema(), selected_indices.iter().copied());
2097 let reader =
2098 builder
2099 .with_projection(projection)
2100 .build()
2101 .map_err(|e| DataError::ParseError {
2102 reason: format!("failed to build parquet reader: {e}"),
2103 })?;
2104 let p = headers.len();
2105 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
2106 if open_ms > 100.0 {
2107 log::info!(
2108 "[DATA-LOAD] parquet_open+meta | n_headers={} | n_proj={} | {:.1}ms",
2109 all_headers.len(),
2110 p,
2111 open_ms
2112 );
2113 }
2114
2115 let t_batches = std::time::Instant::now();
2116 let is_string_col = selected_fields
2117 .iter()
2118 .map(|field| arrow_field_is_string(field.data_type()))
2119 .collect::<Vec<_>>();
2120 let forced_numeric_categorical = headers
2121 .iter()
2122 .enumerate()
2123 .map(|(j, header)| !is_string_col[j] && categorical_roles.contains(header.as_str()))
2124 .collect::<Vec<_>>();
2125 let mut values = Array2::<f64>::zeros((total_rows, p));
2126 let mut all_binary = vec![true; p];
2127 let mut saw_numeric = vec![false; p];
2128 let mut categorical_encoders = (0..p)
2129 .map(|j| {
2130 (is_string_col[j] || forced_numeric_categorical[j]).then(CategoricalEncoder::default)
2131 })
2132 .collect::<Vec<_>>();
2133 let mut rows_seen = 0usize;
2134 for batch_result in reader {
2135 let batch = batch_result.map_err(|e| DataError::ParseError {
2136 reason: format!("failed to read parquet record batch: {e}"),
2137 })?;
2138 let n_rows = batch.num_rows();
2139 if rows_seen.saturating_add(n_rows) > total_rows {
2140 return Err(DataError::SchemaMismatch {
2141 reason: "parquet row count changed while reading record batches".to_string(),
2142 });
2143 }
2144
2145 let decoded_columns = values
2146 .slice_mut(s![rows_seen..rows_seen + n_rows, ..])
2147 .axis_iter_mut(Axis(1))
2148 .into_par_iter()
2149 .zip(categorical_encoders.par_iter_mut())
2150 .zip(all_binary.par_iter_mut())
2151 .zip(saw_numeric.par_iter_mut())
2152 .enumerate()
2153 .map(|(j, (((output, encoder), column_all_binary), column_saw_numeric))| {
2154 decode_arrow_batch_column_into(
2155 batch.column(j).as_ref(),
2156 rows_seen,
2157 &headers[j],
2158 is_string_col[j],
2159 output,
2160 encoder.as_mut(),
2161 column_all_binary,
2162 column_saw_numeric,
2163 )
2164 })
2165 .collect::<Vec<_>>();
2166
2167 for decoded in decoded_columns {
2172 decoded?;
2173 }
2174 rows_seen += n_rows;
2175 }
2176
2177 if rows_seen != total_rows {
2178 return Err(DataError::SchemaMismatch {
2179 reason: format!(
2180 "parquet metadata reports {total_rows} rows but record batches yielded {rows_seen}"
2181 ),
2182 });
2183 }
2184 let batches_ms = t_batches.elapsed().as_secs_f64() * 1000.0;
2185 if batches_ms > 100.0 {
2186 log::info!(
2187 "[DATA-LOAD] parquet_batches_decode | n_rows={} | n_cols={} | {:.1}ms",
2188 total_rows,
2189 p,
2190 batches_ms
2191 );
2192 }
2193 let t_schema = std::time::Instant::now();
2194 let mut levels = vec![Vec::<String>::new(); p];
2200 for (j, encoder) in categorical_encoders.into_iter().enumerate() {
2201 if let Some(encoder) = encoder {
2202 let order = if forced_numeric_categorical[j] {
2203 LevelOrder::Canonical
2204 } else {
2205 LevelOrder::Encounter
2206 };
2207 levels[j] = encoder.finish(values.column_mut(j), order);
2208 }
2209 }
2210 let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
2211 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2212 for j in 0..p {
2213 let kind = if is_string_col[j] || forced_numeric_categorical[j] {
2214 ColumnKindTag::Categorical
2215 } else if all_binary[j] && saw_numeric[j] {
2216 ColumnKindTag::Binary
2217 } else {
2218 ColumnKindTag::Continuous
2219 };
2220 column_kinds.push(kind);
2221 schema_cols.push(SchemaColumn {
2222 name: headers[j].clone(),
2223 kind,
2224 levels: std::mem::take(&mut levels[j]),
2225 });
2226 }
2227 let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
2228 if schema_ms > 100.0 {
2229 let n_cat = column_kinds
2230 .iter()
2231 .filter(|k| matches!(k, ColumnKindTag::Categorical))
2232 .count();
2233 log::info!(
2234 "[DATA-LOAD] parquet_finalize_schema | n_cols={} | n_cat={} | {:.1}ms",
2235 p,
2236 n_cat,
2237 schema_ms
2238 );
2239 }
2240
2241 Ok(EncodedDataset {
2242 headers,
2243 values,
2244 schema: DataSchema {
2245 columns: schema_cols,
2246 },
2247 column_kinds,
2248 })
2249}
2250
2251fn load_parquet_with_schema(
2252 path: &Path,
2253 schema: &DataSchema,
2254 unseen_policy: UnseenCategoryPolicy,
2255 requested_columns: &[String],
2256) -> Result<EncodedDataset, DataError> {
2257 let inferred = load_parquet_inferred(path, requested_columns, &HashSet::new())?;
2261 let p = inferred.headers.len();
2262 let n = inferred.values.nrows();
2263
2264 let schema_byname: HashMap<&str, &SchemaColumn> = schema
2265 .columns
2266 .iter()
2267 .map(|c| (c.name.as_str(), c))
2268 .collect();
2269
2270 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2271 let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
2272 let mut values = inferred.values;
2273
2274 for j in 0..p {
2275 let name = &inferred.headers[j];
2276 if let Some(sc) = schema_byname.get(name.as_str()) {
2277 column_kinds.push(sc.kind);
2278 schema_cols.push((*sc).clone());
2279
2280 match sc.kind {
2281 ColumnKindTag::Continuous => {
2282 if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2283 return Err(DataError::SchemaMismatch {
2284 reason: format!(
2285 "column '{}' is continuous in schema but parquet column is string/categorical",
2286 name
2287 ),
2288 });
2289 }
2290 }
2291 ColumnKindTag::Binary => {
2292 if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2293 return Err(DataError::SchemaMismatch {
2294 reason: format!(
2295 "column '{}' is binary in schema but parquet column is string/categorical",
2296 name
2297 ),
2298 });
2299 }
2300 if let Some(row) = values.column(j).iter().position(|value| {
2302 value.is_finite()
2303 && (*value - 0.0).abs() >= 1e-12
2304 && (*value - 1.0).abs() >= 1e-12
2305 }) {
2306 return Err(DataError::SchemaMismatch {
2307 reason: format!(
2308 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2309 name,
2310 row + 1,
2311 values[[row, j]]
2312 ),
2313 });
2314 }
2315 }
2316 ColumnKindTag::Categorical => {
2317 if !matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2318 return Err(DataError::SchemaMismatch {
2319 reason: format!(
2320 "column '{}' is categorical in schema but parquet column is numeric",
2321 name
2322 ),
2323 });
2324 }
2325 let inferred_col = &inferred.schema.columns[j];
2326 let schema_level_map: HashMap<&str, f64> = sc
2328 .levels
2329 .iter()
2330 .enumerate()
2331 .map(|(idx, v)| (v.as_str(), idx as f64))
2332 .collect();
2333 let inferred_to_schema: Vec<f64> = inferred_col
2334 .levels
2335 .iter()
2336 .map(|lv| {
2337 schema_level_map
2338 .get(lv.as_str())
2339 .copied()
2340 .or_else(|| unseen_policy.unseen_code_for(name, sc.levels.len()))
2341 .ok_or_else(|| DataError::SchemaMismatch {
2342 reason: format!(
2343 "unseen level '{}' in categorical column '{}'",
2344 lv, name
2345 ),
2346 })
2347 })
2348 .collect::<Result<Vec<_>, _>>()?;
2349 for i in 0..n {
2350 let old_code = values[[i, j]] as usize;
2351 if old_code >= inferred_to_schema.len() {
2352 let Some(unseen_code) =
2353 unseen_policy.unseen_code_for(name, sc.levels.len())
2354 else {
2355 return Err(DataError::SchemaMismatch {
2356 reason: format!(
2357 "unseen categorical code at row {}, column '{}'",
2358 i + 1,
2359 name
2360 ),
2361 });
2362 };
2363 values[[i, j]] = unseen_code;
2364 continue;
2365 }
2366 values[[i, j]] = inferred_to_schema[old_code];
2367 }
2368 }
2369 }
2370 } else {
2371 column_kinds.push(inferred.column_kinds[j]);
2373 schema_cols.push(inferred.schema.columns[j].clone());
2374 }
2375 }
2376
2377 Ok(EncodedDataset {
2378 headers: inferred.headers,
2379 values,
2380 schema: DataSchema {
2381 columns: schema_cols,
2382 },
2383 column_kinds,
2384 })
2385}
2386
2387pub fn encode_recordswith_inferred_schema(
2388 headers: Vec<String>,
2389 records: Vec<StringRecord>,
2390) -> Result<EncodedDataset, String> {
2391 if records.is_empty() {
2392 return Err(DataError::EmptyInput {
2393 reason: "table data cannot be empty".to_string(),
2394 }
2395 .into());
2396 }
2397 let schema_cols = headers
2403 .par_iter()
2404 .enumerate()
2405 .map(|(j, name)| infer_schema_column(name, &records, j).map_err(String::from))
2406 .collect::<Result<Vec<SchemaColumn>, String>>()?;
2407 let schema = DataSchema {
2408 columns: schema_cols,
2409 };
2410 encode_recordswith_schema(headers, records, &schema, UnseenCategoryPolicy::Error)
2411}
2412
2413pub fn encode_recordswith_schema(
2414 headers: Vec<String>,
2415 records: Vec<StringRecord>,
2416 schema: &DataSchema,
2417 unseen_policy: UnseenCategoryPolicy,
2418) -> Result<EncodedDataset, String> {
2419 let n = records.len();
2420 if n == 0 {
2421 return Err(DataError::EmptyInput {
2422 reason: "table data cannot be empty".to_string(),
2423 }
2424 .into());
2425 }
2426 let p = headers.len();
2427 if p == 0 {
2428 return Err(DataError::EmptyInput {
2429 reason: "table data must have at least one header column".to_string(),
2430 }
2431 .into());
2432 }
2433 for (i, rec) in records.iter().enumerate() {
2440 if rec.len() != p {
2441 return Err(DataError::SchemaMismatch {
2442 reason: format!(
2443 "row width mismatch at row {}: got {} fields, expected {} (one per header)",
2444 i + 1,
2445 rec.len(),
2446 p
2447 ),
2448 }
2449 .into());
2450 }
2451 }
2452 let schema_byname: HashMap<&str, &SchemaColumn> = schema
2453 .columns
2454 .iter()
2455 .map(|c| (c.name.as_str(), c))
2456 .collect();
2457
2458 let encoded_columns = headers
2464 .par_iter()
2465 .enumerate()
2466 .map(|(j, name)| {
2467 let inferred_for_extra;
2468 let col_schema = if let Some(s) = schema_byname.get(name.as_str()) {
2469 *s
2470 } else {
2471 inferred_for_extra =
2472 infer_schema_column(name, &records, j).map_err(String::from)?;
2473 &inferred_for_extra
2474 };
2475 let column = encode_one_column(name, &records, j, col_schema, &unseen_policy)?;
2476 Ok::<(ColumnKindTag, Vec<f64>), String>((col_schema.kind, column))
2477 })
2478 .collect::<Result<Vec<(ColumnKindTag, Vec<f64>)>, String>>()?;
2479
2480 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2481 let mut values = Array2::<f64>::zeros((n, p));
2482 for (j, (kind, column)) in encoded_columns.into_iter().enumerate() {
2483 column_kinds.push(kind);
2484 values
2485 .column_mut(j)
2486 .assign(&ndarray::ArrayView1::from(&column));
2487 }
2488
2489 Ok(EncodedDataset {
2490 headers,
2491 values,
2492 schema: schema.clone(),
2493 column_kinds,
2494 })
2495}
2496
2497fn encode_one_column(
2504 name: &str,
2505 records: &[StringRecord],
2506 j: usize,
2507 col_schema: &SchemaColumn,
2508 unseen_policy: &UnseenCategoryPolicy,
2509) -> Result<Vec<f64>, String> {
2510 let level_map = if matches!(col_schema.kind, ColumnKindTag::Categorical) {
2511 Some(
2512 col_schema
2513 .levels
2514 .iter()
2515 .enumerate()
2516 .map(|(idx, v)| (v.as_str(), idx as f64))
2517 .collect::<HashMap<_, _>>(),
2518 )
2519 } else {
2520 None
2521 };
2522
2523 let mut column = Vec::<f64>::with_capacity(records.len());
2524 for (i, rec) in records.iter().enumerate() {
2525 let raw = rec
2526 .get(j)
2527 .ok_or_else(|| {
2528 String::from(DataError::SchemaMismatch {
2529 reason: format!("missing field at row {}, col {}", i + 1, j + 1),
2530 })
2531 })?
2532 .trim();
2533 if raw.is_empty() {
2534 return Err(DataError::EmptyInput {
2535 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2536 }
2537 .into());
2538 }
2539 let val = match col_schema.kind {
2540 ColumnKindTag::Continuous if is_missing_marker(raw) => f64::NAN,
2542 ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
2543 String::from(DataError::SchemaMismatch {
2544 reason: format!(
2545 "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
2546 name,
2547 i + 1,
2548 raw,
2549 err
2550 ),
2551 })
2552 })?,
2553 ColumnKindTag::Binary if is_missing_marker(raw) => f64::NAN,
2554 ColumnKindTag::Binary => {
2555 let v = raw.parse::<f64>().map_err(|err| {
2556 String::from(DataError::SchemaMismatch {
2557 reason: format!(
2558 "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
2559 name,
2560 i + 1,
2561 raw,
2562 err
2563 ),
2564 })
2565 })?;
2566 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2567 return Err(DataError::SchemaMismatch {
2568 reason: format!(
2569 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2570 name,
2571 i + 1,
2572 v
2573 ),
2574 }
2575 .into());
2576 }
2577 v
2578 }
2579 ColumnKindTag::Categorical => {
2580 let map = level_map.as_ref().ok_or_else(|| {
2581 String::from(DataError::EncodingFailure {
2582 reason: "internal categorical schema map missing".to_string(),
2583 })
2584 })?;
2585 match map.get(raw) {
2586 Some(v) => *v,
2587 None => unseen_policy
2588 .unseen_code_for(name, col_schema.levels.len())
2589 .ok_or_else(|| {
2590 String::from(DataError::SchemaMismatch {
2591 reason: format!(
2592 "unseen level '{}' in categorical column '{}' at row {}; allowed levels: {}",
2593 raw,
2594 name,
2595 i + 1,
2596 col_schema.levels.join(",")
2597 ),
2598 })
2599 })?,
2600 }
2601 }
2602 };
2603 if !val.is_finite() && !is_missing_marker(raw) {
2605 return Err(DataError::InvalidValue {
2606 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2607 }
2608 .into());
2609 }
2610 column.push(val);
2611 }
2612 Ok(column)
2613}
2614
2615fn infer_schema_column(
2616 name: &str,
2617 records: &[StringRecord],
2618 col_idx: usize,
2619) -> Result<SchemaColumn, DataError> {
2620 let mut all_numeric = true;
2621 let mut all_binary = true;
2622 let mut saw_numeric = false;
2623 let mut levels = Vec::<String>::new();
2624 let mut level_index = HashMap::<String, usize>::new();
2625 let mut missing_markers = Vec::<String>::new();
2629 for (i, rec) in records.iter().enumerate() {
2630 let raw = rec
2631 .get(col_idx)
2632 .ok_or_else(|| DataError::SchemaMismatch {
2633 reason: format!("missing field at row {}, col {}", i + 1, col_idx + 1),
2634 })?
2635 .trim();
2636 if raw.is_empty() {
2637 return Err(DataError::EmptyInput {
2638 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2639 });
2640 }
2641 if is_missing_marker(raw) {
2642 missing_markers.push(raw.to_string());
2643 continue;
2644 }
2645 if let Ok(v) = raw.parse::<f64>() {
2646 saw_numeric = true;
2647 if !v.is_finite() {
2648 return Err(DataError::InvalidValue {
2649 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2650 });
2651 }
2652 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2653 all_binary = false;
2654 }
2655 } else {
2656 all_numeric = false;
2657 all_binary = false;
2658 level_index.entry(raw.to_string()).or_insert_with(|| {
2659 let idx = levels.len();
2660 levels.push(raw.to_string());
2661 idx
2662 });
2663 }
2664 }
2665 let numeric_column = all_numeric && saw_numeric;
2678 if !numeric_column {
2679 for marker in missing_markers {
2680 level_index.entry(marker.clone()).or_insert_with(|| {
2681 let idx = levels.len();
2682 levels.push(marker);
2683 idx
2684 });
2685 }
2686 }
2687 let kind = if numeric_column {
2688 if all_binary {
2689 ColumnKindTag::Binary
2690 } else {
2691 ColumnKindTag::Continuous
2692 }
2693 } else {
2694 ColumnKindTag::Categorical
2695 };
2696 if matches!(kind, ColumnKindTag::Categorical) {
2701 sort_levels_canonical(&mut levels);
2702 }
2703 Ok(SchemaColumn {
2704 name: name.to_string(),
2705 kind,
2706 levels: if matches!(kind, ColumnKindTag::Categorical) {
2707 levels
2708 } else {
2709 Vec::new()
2710 },
2711 })
2712}
2713
2714pub fn infer_and_encode_column_major(
2727 name: &str,
2728 column: &[&str],
2729 col_index: usize,
2730) -> Result<(SchemaColumn, Vec<f64>), String> {
2731 if column.is_empty() {
2732 return Err(DataError::EmptyInput {
2733 reason: "table data cannot be empty".to_string(),
2734 }
2735 .into());
2736 }
2737 let force_categorical = column.iter().any(|c| strip_categorical_sentinel(c).1);
2742 let mut all_numeric = !force_categorical;
2743 let mut all_binary = !force_categorical;
2744 let mut levels = Vec::<String>::new();
2745 let mut level_index = HashMap::<String, usize>::new();
2746 let mut trimmed = Vec::<&str>::with_capacity(column.len());
2747 let mut parsed = Vec::<Option<f64>>::with_capacity(column.len());
2754 let mut saw_numeric = false;
2755 let mut missing_positions = Vec::<usize>::new();
2756 for (i, raw_field) in column.iter().enumerate() {
2757 let (raw, _) = strip_categorical_sentinel(raw_field);
2760 let raw = raw.trim();
2761 if raw.is_empty() {
2762 return Err(DataError::EmptyInput {
2763 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2764 }
2765 .into());
2766 }
2767 if !force_categorical {
2770 if is_missing_marker(raw) {
2774 missing_positions.push(i);
2775 parsed.push(Some(f64::NAN));
2776 trimmed.push(raw);
2777 continue;
2778 }
2779 if let Ok(v) = raw.parse::<f64>() {
2780 saw_numeric = true;
2781 if !v.is_finite() {
2782 return Err(DataError::InvalidValue {
2783 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2784 }
2785 .into());
2786 }
2787 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2788 all_binary = false;
2789 }
2790 parsed.push(Some(v));
2791 trimmed.push(raw);
2792 continue;
2793 }
2794 all_numeric = false;
2795 all_binary = false;
2796 }
2797 level_index.entry(raw.to_string()).or_insert_with(|| {
2798 let idx = levels.len();
2799 levels.push(raw.to_string());
2800 idx
2801 });
2802 parsed.push(None);
2803 trimmed.push(raw);
2804 }
2805 let numeric_column = all_numeric && saw_numeric;
2810 if !numeric_column {
2811 for &i in &missing_positions {
2812 let raw = trimmed[i];
2813 level_index.entry(raw.to_string()).or_insert_with(|| {
2814 let idx = levels.len();
2815 levels.push(raw.to_string());
2816 idx
2817 });
2818 parsed[i] = None;
2819 }
2820 }
2821 let kind = if numeric_column {
2822 if all_binary {
2823 ColumnKindTag::Binary
2824 } else {
2825 ColumnKindTag::Continuous
2826 }
2827 } else {
2828 ColumnKindTag::Categorical
2829 };
2830 if matches!(kind, ColumnKindTag::Categorical) {
2844 sort_levels_canonical(&mut levels);
2845 }
2846 let schema = SchemaColumn {
2847 name: name.to_string(),
2848 kind,
2849 levels: if matches!(kind, ColumnKindTag::Categorical) {
2850 levels
2851 } else {
2852 Vec::new()
2853 },
2854 };
2855
2856 let level_map = if matches!(kind, ColumnKindTag::Categorical) {
2857 Some(
2858 schema
2859 .levels
2860 .iter()
2861 .enumerate()
2862 .map(|(idx, v)| (v.as_str(), idx as f64))
2863 .collect::<HashMap<_, _>>(),
2864 )
2865 } else {
2866 None
2867 };
2868
2869 let mut values = Vec::<f64>::with_capacity(trimmed.len());
2870 for (i, raw) in trimmed.iter().enumerate() {
2871 let raw = *raw;
2872 let val = match kind {
2873 ColumnKindTag::Continuous => parsed[i].ok_or_else(|| {
2877 String::from(DataError::EncodingFailure {
2878 reason: format!(
2879 "internal: continuous column '{}' lost its parsed value at row {} (col {})",
2880 name,
2881 i + 1,
2882 col_index
2883 ),
2884 })
2885 })?,
2886 ColumnKindTag::Binary => {
2887 let v = parsed[i].ok_or_else(|| {
2888 String::from(DataError::EncodingFailure {
2889 reason: format!(
2890 "internal: binary column '{}' lost its parsed value at row {} (col {})",
2891 name,
2892 i + 1,
2893 col_index
2894 ),
2895 })
2896 })?;
2897 if v.is_finite() && (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2900 return Err(DataError::SchemaMismatch {
2901 reason: format!(
2902 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2903 name,
2904 i + 1,
2905 v
2906 ),
2907 }
2908 .into());
2909 }
2910 v
2911 }
2912 ColumnKindTag::Categorical => {
2913 let map = level_map.as_ref().ok_or_else(|| {
2914 String::from(DataError::EncodingFailure {
2915 reason: "internal categorical schema map missing".to_string(),
2916 })
2917 })?;
2918 *map.get(raw).ok_or_else(|| {
2919 String::from(DataError::EncodingFailure {
2920 reason: format!(
2921 "internal: level '{}' missing from freshly built map for column '{}' (col {})",
2922 raw, name, col_index
2923 ),
2924 })
2925 })?
2926 }
2927 };
2928 if !val.is_finite() && !is_missing_marker(raw) {
2930 return Err(DataError::InvalidValue {
2931 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2932 }
2933 .into());
2934 }
2935 values.push(val);
2936 }
2937 Ok((schema, values))
2938}
2939#[cfg(test)]
2940mod missing_value_inference_tests {
2941 use super::*;
2942
2943 fn rows(cells: &[&[&str]]) -> Vec<StringRecord> {
2944 cells
2945 .iter()
2946 .map(|r| StringRecord::from(r.to_vec()))
2947 .collect()
2948 }
2949
2950 #[test]
2951 fn dtype_categorical_missing_cell_is_not_promoted_to_a_level() {
2952 let column = [Some("g10"), None, Some("g2"), Some("g10")];
2953 let (schema, values) = encode_optional_categorical_column("group", &column)
2954 .expect("encode typed categorical values with a missing cell");
2955
2956 assert_eq!(schema.kind, ColumnKindTag::Categorical);
2957 assert_eq!(schema.levels, vec!["g2", "g10"]);
2958 assert_eq!(values[0], 1.0);
2959 assert!(values[1].is_nan());
2960 assert_eq!(values[2], 0.0);
2961 assert_eq!(values[3], 1.0);
2962 }
2963
2964 #[test]
2971 fn a_numeric_column_containing_na_can_never_infer_categorical() {
2972 let ds = encode_recordswith_inferred_schema(
2973 vec!["parker".to_string()],
2974 rows(&[&["94.4"], &["NA"], &["65.0"], &["88.0"], &["NA"]]),
2975 )
2976 .expect("encode a numeric column carrying NA");
2977
2978 assert_eq!(
2979 ds.schema.columns[0].kind,
2980 ColumnKindTag::Continuous,
2981 "a column whose present cells are all numeric is numeric-with-missing, \
2982 never a factor over its own measurements"
2983 );
2984 assert!(
2985 ds.schema.columns[0].levels.is_empty(),
2986 "measurements must not be recorded as factor levels"
2987 );
2988
2989 let col: Vec<f64> = ds.values.column(0).to_vec();
2992 assert_eq!(col[0], 94.4);
2993 assert_eq!(col[2], 65.0);
2994 assert_eq!(col[3], 88.0);
2995 assert!(col[1].is_nan() && col[4].is_nan(), "NA must encode as NaN");
2996 assert_eq!(
2997 col.iter().filter(|v| v.is_finite()).count(),
2998 3,
2999 "is_finite() must count exactly the present cells"
3000 );
3001 }
3002
3003 #[test]
3008 fn na_stays_a_level_in_a_genuinely_categorical_column() {
3009 let ds = encode_recordswith_inferred_schema(
3010 vec!["country".to_string()],
3011 rows(&[&["NA"], &["ZA"], &["BW"], &["NA"]]),
3012 )
3013 .expect("encode a categorical column whose labels include NA");
3014
3015 assert_eq!(ds.schema.columns[0].kind, ColumnKindTag::Categorical);
3016 assert!(
3017 ds.schema.columns[0].levels.iter().any(|l| l == "NA"),
3018 "NA is a country here, not missingness: levels were {:?}",
3019 ds.schema.columns[0].levels
3020 );
3021 assert!(
3022 ds.values.column(0).iter().all(|v| v.is_finite()),
3023 "a categorical column carries level codes, never NaN"
3024 );
3025 }
3026
3027 #[test]
3030 fn a_binary_column_containing_na_stays_binary_with_nan_holes() {
3031 let ds = encode_recordswith_inferred_schema(
3032 vec!["event".to_string()],
3033 rows(&[&["1"], &["NA"], &["0"], &["1"]]),
3034 )
3035 .expect("encode a binary column carrying NA");
3036
3037 assert_eq!(ds.schema.columns[0].kind, ColumnKindTag::Binary);
3038 let col: Vec<f64> = ds.values.column(0).to_vec();
3039 assert_eq!((col[0], col[2], col[3]), (1.0, 0.0, 1.0));
3040 assert!(col[1].is_nan());
3041 }
3042
3043 #[test]
3048 fn a_parsed_non_finite_literal_still_fails_loudly() {
3049 let err = encode_recordswith_inferred_schema(
3050 vec!["x".to_string()],
3051 rows(&[&["1.0"], &["inf"], &["2.0"]]),
3052 )
3053 .expect_err("a literal infinity is a data error, not a missing value");
3054 assert!(
3055 err.contains("non-finite"),
3056 "expected the non-finite guard, got: {err}"
3057 );
3058 }
3059}
3060
3061#[cfg(test)]
3062mod tests {
3063 use super::*;
3064 use arrow::array::ArrayRef;
3065 use arrow::datatypes::{Field, Schema};
3066 use arrow::error::ArrowError;
3067 use arrow::record_batch::{RecordBatch, RecordBatchIterator};
3068 use std::sync::Arc;
3069
3070 fn encode_single_arrow_array(array: ArrayRef) -> Result<EncodedDataset, DataError> {
3071 let schema = Arc::new(Schema::new(vec![Field::new(
3072 "source",
3073 array.data_type().clone(),
3074 true,
3075 )]));
3076 let batch = RecordBatch::try_new(schema.clone(), vec![array]).expect("record batch");
3077 let batches: Vec<Result<RecordBatch, ArrowError>> = vec![Ok(batch)];
3078 let mut reader = RecordBatchIterator::new(batches, schema);
3079 encode_arrow_record_batch_reader_with_inferred_schema(
3080 &mut reader,
3081 vec!["normalized".to_string()],
3082 )
3083 }
3084
3085 #[test]
3086 fn arrow_reader_streams_typed_columns_in_supplied_order() {
3087 use arrow::array::{
3088 Array, BooleanArray, DictionaryArray, Float32Array, Int8Array, Int32Array, Int64Array,
3089 LargeStringArray, StringArray,
3090 };
3091 use arrow::datatypes::Int8Type;
3092
3093 let string_dictionary_1: DictionaryArray<Int8Type> =
3094 vec!["beta", "alpha"].into_iter().collect();
3095 let string_dictionary_2: DictionaryArray<Int8Type> = vec!["beta"].into_iter().collect();
3096 let numeric_dictionary_1 = DictionaryArray::<Int8Type>::new(
3097 Int8Array::from(vec![0, 1]),
3098 Arc::new(Int64Array::from(vec![5, 7])),
3099 );
3100 let numeric_dictionary_2 = DictionaryArray::<Int8Type>::new(
3101 Int8Array::from(vec![0]),
3102 Arc::new(Int64Array::from(vec![5])),
3103 );
3104
3105 let schema = Arc::new(Schema::new(vec![
3106 Field::new("source_float", arrow::datatypes::DataType::Float32, false),
3107 Field::new("source_integer", arrow::datatypes::DataType::Int32, false),
3108 Field::new("source_flag", arrow::datatypes::DataType::Boolean, false),
3109 Field::new("source_utf8", arrow::datatypes::DataType::Utf8, false),
3110 Field::new(
3111 "source_large_utf8",
3112 arrow::datatypes::DataType::LargeUtf8,
3113 false,
3114 ),
3115 Field::new(
3116 "source_dictionary_string",
3117 string_dictionary_1.data_type().clone(),
3118 false,
3119 ),
3120 Field::new(
3121 "source_dictionary_number",
3122 numeric_dictionary_1.data_type().clone(),
3123 false,
3124 ),
3125 ]));
3126 let batch_1 = RecordBatch::try_new(
3127 schema.clone(),
3128 vec![
3129 Arc::new(Float32Array::from(vec![1.5, 2.5])) as ArrayRef,
3130 Arc::new(Int32Array::from(vec![0, 1])),
3131 Arc::new(BooleanArray::from(vec![true, false])),
3132 Arc::new(StringArray::from(vec!["item10", "item2"])),
3133 Arc::new(LargeStringArray::from(vec!["z", "a"])),
3134 Arc::new(string_dictionary_1),
3135 Arc::new(numeric_dictionary_1),
3136 ],
3137 )
3138 .expect("first record batch");
3139 let batch_2 = RecordBatch::try_new(
3140 schema.clone(),
3141 vec![
3142 Arc::new(Float32Array::from(vec![-4.0])) as ArrayRef,
3143 Arc::new(Int32Array::from(vec![1])),
3144 Arc::new(BooleanArray::from(vec![true])),
3145 Arc::new(StringArray::from(vec!["item1"])),
3146 Arc::new(LargeStringArray::from(vec!["z"])),
3147 Arc::new(string_dictionary_2),
3148 Arc::new(numeric_dictionary_2),
3149 ],
3150 )
3151 .expect("second record batch");
3152 let batches: Vec<Result<RecordBatch, ArrowError>> = vec![Ok(batch_1), Ok(batch_2)];
3153 let mut reader = RecordBatchIterator::new(batches, schema);
3154 let headers = [
3155 "float",
3156 "integer",
3157 "flag",
3158 "utf8",
3159 "large_utf8",
3160 "dictionary_string",
3161 "dictionary_number",
3162 ]
3163 .map(str::to_string)
3164 .to_vec();
3165
3166 let dataset =
3167 encode_arrow_record_batch_reader_with_inferred_schema(&mut reader, headers.clone())
3168 .expect("Arrow stream should encode");
3169
3170 assert_eq!(dataset.headers, headers);
3171 assert_eq!(
3172 dataset.column_kinds,
3173 vec![
3174 ColumnKindTag::Continuous,
3175 ColumnKindTag::Binary,
3176 ColumnKindTag::Binary,
3177 ColumnKindTag::Categorical,
3178 ColumnKindTag::Categorical,
3179 ColumnKindTag::Categorical,
3180 ColumnKindTag::Continuous,
3181 ]
3182 );
3183 assert_eq!(
3184 dataset.values,
3185 ndarray::arr2(&[
3186 [1.5, 0.0, 1.0, 2.0, 1.0, 1.0, 5.0],
3187 [2.5, 1.0, 0.0, 1.0, 0.0, 0.0, 7.0],
3188 [-4.0, 1.0, 1.0, 0.0, 1.0, 1.0, 5.0],
3189 ])
3190 );
3191 assert_eq!(
3192 dataset.schema.columns[3].levels,
3193 vec!["item1", "item2", "item10"]
3194 );
3195 assert_eq!(dataset.schema.columns[4].levels, vec!["a", "z"]);
3196 assert_eq!(dataset.schema.columns[5].levels, vec!["alpha", "beta"]);
3197 assert!(
3198 dataset
3199 .schema
3200 .columns
3201 .iter()
3202 .zip(dataset.headers.iter())
3203 .all(|(column, header)| column.name == *header)
3204 );
3205 }
3206
3207 #[test]
3208 fn arrow_reader_rejects_empty_duplicate_and_mismatched_headers() {
3209 let schema = Arc::new(Schema::new(vec![Field::new(
3210 "source",
3211 arrow::datatypes::DataType::Int32,
3212 false,
3213 )]));
3214
3215 let mut empty_name_reader = RecordBatchIterator::new(
3216 Vec::<Result<RecordBatch, ArrowError>>::new(),
3217 schema.clone(),
3218 );
3219 let empty_name = encode_arrow_record_batch_reader_with_inferred_schema(
3220 &mut empty_name_reader,
3221 vec![" ".to_string()],
3222 )
3223 .expect_err("blank header should fail");
3224 assert!(matches!(empty_name, DataError::EmptyInput { .. }));
3225
3226 let mut duplicate_reader = RecordBatchIterator::new(
3227 Vec::<Result<RecordBatch, ArrowError>>::new(),
3228 Arc::new(Schema::new(vec![
3229 Field::new("a", arrow::datatypes::DataType::Int32, false),
3230 Field::new("b", arrow::datatypes::DataType::Int32, false),
3231 ])),
3232 );
3233 let duplicate = encode_arrow_record_batch_reader_with_inferred_schema(
3234 &mut duplicate_reader,
3235 vec!["x".to_string(), "x".to_string()],
3236 )
3237 .expect_err("duplicate header should fail");
3238 assert!(matches!(duplicate, DataError::SchemaMismatch { .. }));
3239
3240 let mut mismatch_reader =
3241 RecordBatchIterator::new(Vec::<Result<RecordBatch, ArrowError>>::new(), schema);
3242 let mismatch = encode_arrow_record_batch_reader_with_inferred_schema(
3243 &mut mismatch_reader,
3244 vec!["x".to_string(), "y".to_string()],
3245 )
3246 .expect_err("header count mismatch should fail");
3247 assert!(matches!(mismatch, DataError::SchemaMismatch { .. }));
3248 }
3249
3250 #[test]
3251 fn arrow_reader_preserves_missing_cells_and_rejects_unsupported_types() {
3252 use arrow::array::{Date32Array, DictionaryArray, Float64Array, Int8Array, StringArray};
3253 use arrow::datatypes::Int8Type;
3254
3255 let null_numeric =
3256 encode_single_arrow_array(Arc::new(Float64Array::from(vec![Some(1.0), None])))
3257 .expect("numeric null should remain representable until model projection");
3258 assert_eq!(null_numeric.values[[0, 0]], 1.0);
3259 assert!(null_numeric.values[[1, 0]].is_nan());
3260
3261 let null_dictionary = DictionaryArray::<Int8Type>::new(
3262 Int8Array::from(vec![0, 1, 2]),
3263 Arc::new(StringArray::from(vec![Some("present"), Some(""), None])),
3264 );
3265 let logical_null = encode_single_arrow_array(Arc::new(null_dictionary))
3266 .expect("empty and null dictionary values should remain representable");
3267 assert_eq!(logical_null.schema.columns[0].levels, vec!["present"]);
3268 assert_eq!(logical_null.values[[0, 0]], 0.0);
3269 assert!(logical_null.values[[1, 0]].is_nan());
3270 assert!(logical_null.values[[2, 0]].is_nan());
3271
3272 let nonfinite = encode_single_arrow_array(Arc::new(Float64Array::from(vec![
3273 f64::NAN,
3274 f64::INFINITY,
3275 f64::NEG_INFINITY,
3276 ])))
3277 .expect("non-finite values should remain representable until model projection");
3278 assert!(nonfinite.values.column(0).iter().all(|value| value.is_nan()));
3279 assert_eq!(
3280 nonfinite.column_kinds,
3281 vec![ColumnKindTag::Continuous],
3282 "an all-missing typed numeric column is not vacuously binary"
3283 );
3284
3285 let unsupported = encode_single_arrow_array(Arc::new(Date32Array::from(vec![1])))
3286 .expect_err("date column should fail");
3287 assert!(matches!(&unsupported, DataError::InvalidValue { .. }));
3288 assert!(
3289 unsupported
3290 .to_string()
3291 .contains("unsupported Arrow column type")
3292 );
3293 }
3294
3295 #[test]
3296 fn encode_records_rejects_empty_input() {
3297 let headers = vec!["x".to_string()];
3298 let schema = DataSchema {
3299 columns: vec![SchemaColumn {
3300 name: "x".to_string(),
3301 kind: ColumnKindTag::Continuous,
3302 levels: Vec::new(),
3303 }],
3304 };
3305
3306 let err = encode_recordswith_inferred_schema(headers.clone(), Vec::new())
3307 .expect_err("empty inferred records should error");
3308 assert_eq!(err, "table data cannot be empty");
3309
3310 let err =
3311 encode_recordswith_schema(headers, Vec::new(), &schema, UnseenCategoryPolicy::Error)
3312 .expect_err("empty schema-guided records should error");
3313 assert_eq!(err, "table data cannot be empty");
3314 }
3315
3316 #[test]
3317 fn column_major_matches_record_driven_inferred_encode() {
3318 let headers = vec!["cont".to_string(), "bin".to_string(), "cat".to_string()];
3323 let raw_rows = vec![
3324 vec!["1.5", "0", "a"],
3325 vec!["2.0", "1", "b"],
3326 vec!["-3.25", "1", "a"],
3327 vec!["0.0", "0", "c"],
3328 ];
3329 let records: Vec<StringRecord> = raw_rows
3330 .iter()
3331 .map(|r| StringRecord::from(r.clone()))
3332 .collect();
3333 let record_ds = encode_recordswith_inferred_schema(headers.clone(), records)
3334 .expect("record-driven encode");
3335
3336 for (j, name) in headers.iter().enumerate() {
3337 let column: Vec<&str> = raw_rows.iter().map(|r| r[j]).collect();
3338 let (schema_col, values) =
3339 infer_and_encode_column_major(name, &column, j + 1).expect("column-major encode");
3340 assert_eq!(schema_col.kind, record_ds.schema.columns[j].kind);
3341 assert_eq!(schema_col.levels, record_ds.schema.columns[j].levels);
3342 for (i, v) in values.iter().enumerate() {
3343 assert_eq!(*v, record_ds.values[[i, j]], "row {i} col {name}");
3344 }
3345 }
3346 }
3347
3348 #[test]
3349 fn encode_records_can_encode_unseen_named_categorical_column() {
3350 let schema = DataSchema {
3351 columns: vec![
3352 SchemaColumn {
3353 name: "g".to_string(),
3354 kind: ColumnKindTag::Categorical,
3355 levels: vec!["a".to_string(), "b".to_string()],
3356 },
3357 SchemaColumn {
3358 name: "x".to_string(),
3359 kind: ColumnKindTag::Categorical,
3360 levels: vec!["low".to_string(), "high".to_string()],
3361 },
3362 ],
3363 };
3364 let headers = vec!["g".to_string(), "x".to_string()];
3365 let records = vec![StringRecord::from(vec!["new-group", "low"])];
3366 let policy =
3367 UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
3368
3369 let ds =
3370 encode_recordswith_schema(headers, records, &schema, policy).expect("encoded dataset");
3371
3372 assert_eq!(ds.values[[0, 0]], 2.0);
3373 assert_eq!(ds.values[[0, 1]], 0.0);
3374 }
3375
3376 #[test]
3377 fn categorical_encoder_consumes_labels_and_remaps_canonically() {
3378 use ndarray::Array1;
3379
3380 let mut encoder = CategoricalEncoder::default();
3381 let mut encoded = Array1::from_vec(
3382 ["item10", "item2", "item1", "item2"]
3383 .into_iter()
3384 .map(|label| encoder.encode(label) as f64)
3385 .collect(),
3386 );
3387
3388 let levels = encoder.finish(encoded.view_mut(), LevelOrder::Canonical);
3389
3390 assert_eq!(levels, vec!["item1", "item2", "item10"]);
3391 assert_eq!(encoded.to_vec(), vec![2.0, 1.0, 0.0, 1.0]);
3392 }
3393
3394 #[test]
3395 fn complete_delimited_schema_encodes_projected_rows_directly() {
3396 let dir = tempfile::tempdir().expect("tempdir");
3397 let path = dir.path().join("schema_direct.csv");
3398 std::fs::write(
3399 &path,
3400 "y,group,flag,unused\n1.5,b,0,first\n2.5,a,1,second\n",
3401 )
3402 .expect("write csv");
3403 let schema = DataSchema {
3404 columns: vec![
3405 SchemaColumn {
3406 name: "group".to_string(),
3407 kind: ColumnKindTag::Categorical,
3408 levels: vec!["a".to_string(), "b".to_string()],
3409 },
3410 SchemaColumn {
3411 name: "flag".to_string(),
3412 kind: ColumnKindTag::Binary,
3413 levels: Vec::new(),
3414 },
3415 SchemaColumn {
3416 name: "y".to_string(),
3417 kind: ColumnKindTag::Continuous,
3418 levels: Vec::new(),
3419 },
3420 ],
3421 };
3422
3423 let loaded = load_datasetwith_schema_projected(
3424 &path,
3425 &schema,
3426 UnseenCategoryPolicy::Error,
3427 &["y".to_string(), "group".to_string(), "flag".to_string()],
3428 )
3429 .expect("schema-guided projected load");
3430
3431 assert_eq!(loaded.headers, vec!["y", "group", "flag"]);
3432 assert_eq!(loaded.values.row(0).to_vec(), vec![1.5, 1.0, 0.0]);
3433 assert_eq!(loaded.values.row(1).to_vec(), vec![2.5, 0.0, 1.0]);
3434 assert_eq!(
3435 loaded.column_kinds,
3436 vec![
3437 ColumnKindTag::Continuous,
3438 ColumnKindTag::Categorical,
3439 ColumnKindTag::Binary,
3440 ]
3441 );
3442 }
3443
3444 #[test]
3445 fn direct_parquet_decoder_preserves_string_encounter_order() {
3446 use arrow::array::DictionaryArray;
3447 use arrow::datatypes::Int8Type;
3448 use ndarray::Array1;
3449
3450 let dictionary: DictionaryArray<Int8Type> =
3451 vec!["beta", "alpha", "beta"].into_iter().collect();
3452 let mut encoded = Array1::<f64>::zeros(dictionary.len());
3453 let mut encoder = CategoricalEncoder::default();
3454 let mut all_binary = true;
3455 let mut saw_numeric = false;
3456
3457 decode_arrow_batch_column_into(
3458 &dictionary,
3459 0,
3460 "group",
3461 true,
3462 encoded.view_mut(),
3463 Some(&mut encoder),
3464 &mut all_binary,
3465 &mut saw_numeric,
3466 )
3467 .expect("dictionary strings decode directly");
3468 let levels = encoder.finish(encoded.view_mut(), LevelOrder::Encounter);
3469
3470 assert_eq!(levels, vec!["beta", "alpha"]);
3471 assert_eq!(encoded.to_vec(), vec![0.0, 1.0, 0.0]);
3472 }
3473
3474 #[test]
3475 fn numeric_valued_dictionary_column_classifies_and_decodes_as_numeric() {
3476 use arrow::array::{Array, ArrayRef, DictionaryArray, Int8Array, Int64Array};
3486 use arrow::datatypes::{DataType, Int8Type};
3487 use std::sync::Arc;
3488
3489 let keys = Int8Array::from(vec![0i8, 1, 0, 1, 0]);
3491 let dict_values: ArrayRef = Arc::new(Int64Array::from(vec![5i64, 7]));
3492 let dict: DictionaryArray<Int8Type> = DictionaryArray::new(keys, dict_values);
3493
3494 assert!(matches!(dict.data_type(), DataType::Dictionary(_, _)));
3497 assert!(
3498 !arrow_field_is_string(dict.data_type()),
3499 "Dictionary(Int8, Int64) must not be treated as a string column"
3500 );
3501
3502 let str_dict: DictionaryArray<Int8Type> = vec!["a", "b", "a"].into_iter().collect();
3504 assert!(
3505 arrow_field_is_string(str_dict.data_type()),
3506 "Dictionary(Int8, Utf8) must remain a string column"
3507 );
3508
3509 let mut decoded = ndarray::Array1::<f64>::zeros(dict.len());
3513 let mut all_binary = true;
3514 let mut saw_numeric = false;
3515 decode_arrow_batch_column_into(
3516 &dict,
3517 0,
3518 "x",
3519 false,
3520 decoded.view_mut(),
3521 None,
3522 &mut all_binary,
3523 &mut saw_numeric,
3524 )
3525 .expect("numeric dictionary column should decode as numeric");
3526 assert_eq!(decoded.to_vec(), vec![5.0, 7.0, 5.0, 7.0, 5.0]);
3527 assert!(!all_binary);
3528
3529 use arrow::datatypes::{Field, Schema};
3534 use arrow::record_batch::RecordBatch;
3535 use parquet::arrow::ArrowWriter;
3536
3537 let arrow_schema = Arc::new(Schema::new(vec![Field::new(
3538 "x",
3539 dict.data_type().clone(),
3540 false,
3541 )]));
3542 let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(dict.clone())])
3543 .expect("record batch with a dictionary numeric column");
3544
3545 let dir = tempfile::tempdir().expect("tempdir");
3546 let path = dir.path().join("dict_numeric.parquet");
3547 {
3548 let file = std::fs::File::create(&path).expect("create parquet");
3549 let mut writer =
3550 ArrowWriter::try_new(file, arrow_schema, None).expect("arrow parquet writer");
3551 writer.write(&batch).expect("write batch");
3552 writer.close().expect("close writer");
3553 }
3554
3555 let inferred =
3558 load_parquet_inferred(&path, &[], &HashSet::new()).expect("inferred parquet load");
3559 assert_eq!(inferred.column_kinds, vec![ColumnKindTag::Continuous]);
3560 assert_eq!(
3561 inferred.values.column(0).to_vec(),
3562 vec![5.0, 7.0, 5.0, 7.0, 5.0]
3563 );
3564
3565 let schema = DataSchema {
3569 columns: vec![SchemaColumn {
3570 name: "x".to_string(),
3571 kind: ColumnKindTag::Continuous,
3572 levels: Vec::new(),
3573 }],
3574 };
3575 let schema_loaded =
3576 load_parquet_with_schema(&path, &schema, UnseenCategoryPolicy::Error, &[])
3577 .expect("dictionary-encoded numeric parquet must load against a Continuous schema");
3578 assert_eq!(schema_loaded.column_kinds, vec![ColumnKindTag::Continuous]);
3579 assert_eq!(
3580 schema_loaded.values.column(0).to_vec(),
3581 vec![5.0, 7.0, 5.0, 7.0, 5.0]
3582 );
3583 }
3584
3585 #[test]
3586 fn encode_records_keeps_unlisted_categorical_columns_strict() {
3587 let schema = DataSchema {
3588 columns: vec![
3589 SchemaColumn {
3590 name: "g".to_string(),
3591 kind: ColumnKindTag::Categorical,
3592 levels: vec!["a".to_string(), "b".to_string()],
3593 },
3594 SchemaColumn {
3595 name: "x".to_string(),
3596 kind: ColumnKindTag::Categorical,
3597 levels: vec!["low".to_string(), "high".to_string()],
3598 },
3599 ],
3600 };
3601 let headers = vec!["g".to_string(), "x".to_string()];
3602 let records = vec![StringRecord::from(vec!["a", "new-level"])];
3603 let policy =
3604 UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
3605
3606 let err = encode_recordswith_schema(headers, records, &schema, policy)
3607 .expect_err("ordinary categorical column should stay strict");
3608
3609 assert!(err.contains("unseen level 'new-level' in categorical column 'x'"));
3610 }
3611
3612 #[test]
3617 fn sentinel_strip_present_returns_rest_and_true() {
3618 let marked = format!("{}{}", CATEGORICAL_CELL_SENTINEL, "hello");
3619 let (rest, found) = strip_categorical_sentinel(&marked);
3620 assert_eq!(rest, "hello");
3621 assert!(found);
3622 }
3623
3624 #[test]
3625 fn sentinel_strip_absent_returns_original_and_false() {
3626 let (rest, found) = strip_categorical_sentinel("hello");
3627 assert_eq!(rest, "hello");
3628 assert!(!found);
3629 }
3630
3631 #[test]
3632 fn sentinel_strip_empty_string_returns_empty_and_false() {
3633 let (rest, found) = strip_categorical_sentinel("");
3634 assert_eq!(rest, "");
3635 assert!(!found);
3636 }
3637
3638 #[test]
3639 fn sentinel_strip_only_sentinel_returns_empty_and_true() {
3640 let marked = CATEGORICAL_CELL_SENTINEL.to_string();
3641 let (rest, found) = strip_categorical_sentinel(&marked);
3642 assert_eq!(rest, "");
3643 assert!(found);
3644 }
3645
3646 #[test]
3651 fn feature_ranges_two_columns() {
3652 let values = ndarray::arr2(&[[1.0_f64, 10.0], [3.0, 20.0], [2.0, 15.0]]);
3653 let ds = EncodedDataset {
3654 headers: vec!["a".to_string(), "b".to_string()],
3655 values,
3656 schema: DataSchema { columns: vec![] },
3657 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3658 };
3659 let ranges = ds.feature_ranges();
3660 assert_eq!(ranges.len(), 2);
3661 assert_eq!(ranges[0], (1.0, 3.0));
3662 assert_eq!(ranges[1], (10.0, 20.0));
3663 }
3664
3665 #[test]
3666 fn feature_ranges_single_row_min_equals_max() {
3667 let values = ndarray::arr2(&[[5.0_f64, -3.0]]);
3668 let ds = EncodedDataset {
3669 headers: vec!["x".to_string(), "y".to_string()],
3670 values,
3671 schema: DataSchema { columns: vec![] },
3672 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3673 };
3674 let ranges = ds.feature_ranges();
3675 assert_eq!(ranges[0], (5.0, 5.0));
3676 assert_eq!(ranges[1], (-3.0, -3.0));
3677 }
3678
3679 #[test]
3680 fn feature_ranges_all_nan_defaults_to_zero() {
3681 let values = ndarray::arr2(&[[f64::NAN], [f64::NAN]]);
3682 let ds = EncodedDataset {
3683 headers: vec!["x".to_string()],
3684 values,
3685 schema: DataSchema { columns: vec![] },
3686 column_kinds: vec![ColumnKindTag::Continuous],
3687 };
3688 let ranges = ds.feature_ranges();
3689 assert_eq!(ranges[0], (0.0, 0.0));
3690 }
3691
3692 #[test]
3697 fn column_map_indexes_by_name() {
3698 let values = ndarray::arr2(&[[0.0_f64, 1.0], [2.0, 3.0]]);
3699 let ds = EncodedDataset {
3700 headers: vec!["alpha".to_string(), "beta".to_string()],
3701 values,
3702 schema: DataSchema { columns: vec![] },
3703 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3704 };
3705 let map = ds.column_map();
3706 assert_eq!(map["alpha"], 0);
3707 assert_eq!(map["beta"], 1);
3708 assert_eq!(map.len(), 2);
3709 }
3710
3711 #[test]
3714 fn shared_prefix_identical_strings() {
3715 assert_eq!(shared_prefix("hello", "hello"), 5);
3716 }
3717
3718 #[test]
3719 fn shared_prefix_no_common_prefix() {
3720 assert_eq!(shared_prefix("abc", "xyz"), 0);
3721 }
3722
3723 #[test]
3724 fn shared_prefix_partial_match() {
3725 assert_eq!(shared_prefix("foobar", "foobaz"), 5);
3726 }
3727
3728 #[test]
3729 fn shared_prefix_one_empty() {
3730 assert_eq!(shared_prefix("", "hello"), 0);
3731 assert_eq!(shared_prefix("hello", ""), 0);
3732 }
3733
3734 #[test]
3735 fn shared_prefix_both_empty() {
3736 assert_eq!(shared_prefix("", ""), 0);
3737 }
3738
3739 #[test]
3740 fn shared_prefix_shorter_string_is_prefix() {
3741 assert_eq!(shared_prefix("foo", "foobar"), 3);
3742 }
3743
3744 #[test]
3747 fn detect_format_csv() {
3748 let path = std::path::Path::new("data.csv");
3749 assert_eq!(detect_format(path).unwrap(), DataFormat::Csv);
3750 }
3751
3752 #[test]
3753 fn detect_format_tsv() {
3754 assert_eq!(
3755 detect_format(std::path::Path::new("data.tsv")).unwrap(),
3756 DataFormat::Tsv
3757 );
3758 assert_eq!(
3759 detect_format(std::path::Path::new("data.txt")).unwrap(),
3760 DataFormat::Tsv
3761 );
3762 assert_eq!(
3763 detect_format(std::path::Path::new("data.tab")).unwrap(),
3764 DataFormat::Tsv
3765 );
3766 }
3767
3768 #[test]
3769 fn detect_format_parquet() {
3770 assert_eq!(
3771 detect_format(std::path::Path::new("data.parquet")).unwrap(),
3772 DataFormat::Parquet
3773 );
3774 assert_eq!(
3775 detect_format(std::path::Path::new("data.pq")).unwrap(),
3776 DataFormat::Parquet
3777 );
3778 assert_eq!(
3779 detect_format(std::path::Path::new("data.pqt")).unwrap(),
3780 DataFormat::Parquet
3781 );
3782 }
3783
3784 #[test]
3785 fn detect_format_uppercase_extension() {
3786 assert_eq!(
3787 detect_format(std::path::Path::new("data.CSV")).unwrap(),
3788 DataFormat::Csv
3789 );
3790 }
3791
3792 #[test]
3793 fn detect_format_unknown_extension_is_error() {
3794 let err = detect_format(std::path::Path::new("data.json")).unwrap_err();
3795 let msg = format!("{err:?}");
3796 assert!(
3797 msg.contains("json") || msg.contains("unsupported"),
3798 "error should mention extension, got: {msg}"
3799 );
3800 }
3801
3802 #[test]
3805 fn strip_categorical_sentinel_marked_cell() {
3806 let marked = "\u{0}hello";
3808 let (text, found) = strip_categorical_sentinel(marked);
3809 assert!(found);
3810 assert_eq!(text, "hello");
3811 }
3812
3813 #[test]
3814 fn strip_categorical_sentinel_unmarked_cell() {
3815 let (text, found) = strip_categorical_sentinel("plain");
3816 assert!(!found);
3817 assert_eq!(text, "plain");
3818 }
3819
3820 #[test]
3821 fn strip_categorical_sentinel_empty_string() {
3822 let (text, found) = strip_categorical_sentinel("");
3823 assert!(!found);
3824 assert_eq!(text, "");
3825 }
3826
3827 #[test]
3828 fn strip_categorical_sentinel_only_sentinel() {
3829 let s = "\u{0}";
3830 let (text, found) = strip_categorical_sentinel(s);
3831 assert!(found);
3832 assert_eq!(text, "");
3833 }
3834
3835 #[test]
3838 fn projected_headers_selects_by_index() {
3839 let all = vec![
3840 "a".to_string(),
3841 "b".to_string(),
3842 "c".to_string(),
3843 "d".to_string(),
3844 ];
3845 let selected = projected_headers(&all, &[1, 3]);
3846 assert_eq!(selected, vec!["b".to_string(), "d".to_string()]);
3847 }
3848
3849 #[test]
3850 fn projected_headers_empty_selection() {
3851 let all = vec!["x".to_string(), "y".to_string()];
3852 let selected = projected_headers(&all, &[]);
3853 assert!(selected.is_empty());
3854 }
3855
3856 #[test]
3857 fn projected_headers_all_indices() {
3858 let all = vec!["p".to_string(), "q".to_string()];
3859 let selected = projected_headers(&all, &[0, 1]);
3860 assert_eq!(selected, all);
3861 }
3862
3863 #[test]
3864 fn canonical_level_bits_collapses_signed_zero() {
3865 let pos = 0.0_f64;
3869 let neg = -0.0_f64;
3870 assert_ne!(
3871 pos.to_bits(),
3872 neg.to_bits(),
3873 "precondition: raw bits differ"
3874 );
3875 assert_eq!(pos, neg, "precondition: numerically equal");
3876 assert_eq!(canonical_level_bits(pos), canonical_level_bits(neg));
3877 assert_eq!(canonical_level_bits(neg), 0.0_f64.to_bits());
3878 assert_eq!(canonical_level_bits(-1.0 * 0.0), 0.0_f64.to_bits());
3880 assert_eq!(canonical_level_bits(0.0 - 0.0), 0.0_f64.to_bits());
3881 }
3882
3883 #[test]
3884 fn canonical_level_bits_is_bit_stable_on_ordinary_values() {
3885 for &v in &[
3888 1.0_f64,
3889 -1.0,
3890 2.5,
3891 -3.75,
3892 1e300,
3893 -1e-300,
3894 f64::MIN,
3895 f64::MAX,
3896 ] {
3897 assert_eq!(canonical_level_bits(v), v.to_bits(), "value {v}");
3898 }
3899 assert_ne!(canonical_level_bits(1.0), canonical_level_bits(2.0));
3901 assert_ne!(canonical_level_bits(0.0), canonical_level_bits(1.0));
3902 assert_ne!(
3904 canonical_level_bits(f64::INFINITY),
3905 canonical_level_bits(f64::NEG_INFINITY)
3906 );
3907 }
3908
3909 #[test]
3910 fn canonical_level_bits_collapses_nan_payloads() {
3911 let a = f64::NAN;
3913 let b = f64::from_bits(0x7ff8_0000_0000_0001); let c = -f64::NAN; assert!(a.is_nan() && b.is_nan() && c.is_nan());
3916 assert_eq!(canonical_level_bits(a), canonical_level_bits(b));
3917 assert_eq!(canonical_level_bits(a), canonical_level_bits(c));
3918 }
3919
3920 #[test]
3921 fn canonical_level_bits_is_idempotent() {
3922 for &v in &[0.0_f64, -0.0, 1.0, -2.0, f64::NAN] {
3925 let once = canonical_level_bits(v);
3926 let twice = canonical_level_bits(f64::from_bits(once));
3927 assert_eq!(once, twice, "value {v}");
3928 }
3929 }
3930 #[test]
3931 fn fit_boundary_reports_each_degenerate_column_by_name() {
3932 let cases = [
3933 (
3934 vec![0.0, f64::NAN, 1.0],
3935 "has non-finite value NaN at row 2",
3936 ),
3937 (
3938 vec![0.0, f64::INFINITY, 1.0],
3939 "has non-finite value inf at row 2",
3940 ),
3941 (
3942 vec![0.0, f64::NEG_INFINITY, 1.0],
3943 "has non-finite value -inf at row 2",
3944 ),
3945 (
3946 vec![f64::NAN, 2.0, f64::NAN],
3947 "has only one non-missing value",
3948 ),
3949 ];
3950 for (values, expected) in cases {
3951 let dataset = EncodedDataset {
3952 headers: vec!["temperature".to_string()],
3953 values: Array2::from_shape_vec((3, 1), values).unwrap(),
3954 schema: DataSchema {
3955 columns: vec![SchemaColumn {
3956 name: "temperature".to_string(),
3957 kind: ColumnKindTag::Continuous,
3958 levels: Vec::new(),
3959 }],
3960 },
3961 column_kinds: vec![ColumnKindTag::Continuous],
3962 };
3963 let error = dataset.validate_fit_boundary().unwrap_err();
3964 assert!(matches!(error, DataError::DegenerateColumn { .. }));
3965 assert_eq!(
3966 error.to_string(),
3967 format!("column 'temperature' {expected}")
3968 );
3969 }
3970 }
3971
3972 #[test]
3973 fn fit_boundary_rejects_empty_duplicate_and_one_level_factor() {
3974 let cases = [
3975 EncodedDataset {
3976 headers: vec!["x".into()],
3977 values: Array2::zeros((0, 1)),
3978 schema: DataSchema {
3979 columns: vec![SchemaColumn {
3980 name: "x".into(),
3981 kind: ColumnKindTag::Continuous,
3982 levels: vec![],
3983 }],
3984 },
3985 column_kinds: vec![ColumnKindTag::Continuous],
3986 },
3987 EncodedDataset {
3988 headers: vec!["x".into(), "x".into()],
3989 values: Array2::from_shape_vec((2, 2), vec![0.0, 1.0, 1.0, 0.0]).unwrap(),
3990 schema: DataSchema { columns: vec![] },
3991 column_kinds: vec![ColumnKindTag::Continuous; 2],
3992 },
3993 EncodedDataset {
3994 headers: vec!["group".into()],
3995 values: Array2::zeros((2, 1)),
3996 schema: DataSchema {
3997 columns: vec![SchemaColumn {
3998 name: "group".into(),
3999 kind: ColumnKindTag::Categorical,
4000 levels: vec!["only".into()],
4001 }],
4002 },
4003 column_kinds: vec![ColumnKindTag::Categorical],
4004 },
4005 ];
4006 let expected = [
4007 "column '<table>' has no observations",
4008 "column 'x' has a duplicate name",
4009 "column 'group' is a factor with fewer than two levels",
4010 ];
4011 for (dataset, expected) in cases.into_iter().zip(expected) {
4012 assert_eq!(
4013 dataset.validate_fit_boundary().unwrap_err().to_string(),
4014 expected
4015 );
4016 }
4017 }
4018}