1use csv::{ReaderBuilder, StringRecord};
2use ndarray::{Array2, ArrayViewMut1, Axis, s};
3use rayon::prelude::*;
4use serde::{Deserialize, Serialize};
5use std::cmp::Ordering;
6use std::collections::{HashMap, HashSet};
7use std::fmt;
8use std::path::Path;
9
10fn natural_level_cmp(a: &str, b: &str) -> Ordering {
11 let mut ia = 0;
12 let mut ib = 0;
13 let ba = a.as_bytes();
14 let bb = b.as_bytes();
15 while ia < ba.len() && ib < bb.len() {
16 if ba[ia].is_ascii_digit() && bb[ib].is_ascii_digit() {
17 let sa = ia;
18 let sb = ib;
19 while ia < ba.len() && ba[ia].is_ascii_digit() {
20 ia += 1;
21 }
22 while ib < bb.len() && bb[ib].is_ascii_digit() {
23 ib += 1;
24 }
25 let da = &a[sa..ia];
26 let db = &b[sb..ib];
27 let ta = da.trim_start_matches('0');
28 let tb = db.trim_start_matches('0');
29 let ta = if ta.is_empty() { "0" } else { ta };
30 let tb = if tb.is_empty() { "0" } else { tb };
31 match ta.len().cmp(&tb.len()).then_with(|| ta.cmp(tb)) {
32 Ordering::Equal if da.len() != db.len() => return da.len().cmp(&db.len()),
33 Ordering::Equal => {}
34 ord => return ord,
35 }
36 } else {
37 match ba[ia].cmp(&bb[ib]) {
38 Ordering::Equal => {
39 ia += 1;
40 ib += 1;
41 }
42 ord => return ord,
43 }
44 }
45 }
46 ba.len().cmp(&bb.len())
47}
48
49fn sort_levels_canonical(levels: &mut [String]) {
50 levels.sort_by(|a, b| natural_level_cmp(a, b));
51}
52
53#[inline]
83pub fn canonical_level_bits(v: f64) -> u64 {
84 if v == 0.0 {
85 0.0_f64.to_bits()
87 } else if v.is_nan() {
88 f64::NAN.to_bits()
90 } else {
91 v.to_bits()
92 }
93}
94
95#[derive(Debug, Clone)]
106pub enum DataError {
107 SchemaMismatch { reason: String },
112 ParseError { reason: String },
116 EncodingFailure { reason: String },
120 EmptyInput { reason: String },
123 InvalidValue { reason: String },
126 ColumnNotFound {
133 name: String,
135 role: Option<String>,
139 available: Vec<String>,
141 similar: Vec<String>,
144 tsv_hint: bool,
149 },
150}
151
152impl DataError {
153 pub fn column_not_found(
159 col_map: &HashMap<String, usize>,
160 name: &str,
161 role: Option<&str>,
162 ) -> Self {
163 let target_lower = name.to_lowercase();
164 let mut similar: Vec<String> = col_map
165 .keys()
166 .filter(|k| {
167 let k_lower = k.to_lowercase();
168 k_lower.contains(&target_lower)
169 || target_lower.contains(&k_lower)
170 || shared_prefix(&k_lower, &target_lower) >= 3
171 })
172 .cloned()
173 .collect();
174 similar.sort_unstable();
175 let mut available: Vec<String> = col_map.keys().cloned().collect();
176 available.sort_unstable();
177 let tsv_hint = available.len() == 1 && available[0].contains('\t');
178 Self::ColumnNotFound {
179 name: name.to_string(),
180 role: role.map(str::to_string),
181 available,
182 similar,
183 tsv_hint,
184 }
185 }
186}
187
188impl fmt::Display for DataError {
189 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
190 match self {
191 DataError::SchemaMismatch { reason }
192 | DataError::ParseError { reason }
193 | DataError::EncodingFailure { reason }
194 | DataError::EmptyInput { reason }
195 | DataError::InvalidValue { reason } => f.write_str(reason),
196 DataError::ColumnNotFound {
197 name,
198 role,
199 available,
200 similar,
201 tsv_hint,
202 } => {
203 let label = match role {
204 Some(r) => format!("{r} column '{name}'"),
205 None => format!("column '{name}'"),
206 };
207 let tsv_suffix = if *tsv_hint {
208 " — your file appears to be tab-separated; gam expects comma-separated CSV. \
209 Replace tabs with commas, or pre-convert with `tr '\\t' ',' < file.tsv > file.csv`."
210 } else {
211 ""
212 };
213 if similar.is_empty() {
214 write!(
215 f,
216 "{label} not found in data. Available columns: [{}]{tsv_suffix}",
217 available.join(", ")
218 )
219 } else {
220 write!(
221 f,
222 "{label} not found in data. Did you mean one of [{}]? Full list: [{}]{tsv_suffix}",
223 similar.join(", "),
224 available.join(", ")
225 )
226 }
227 }
228 }
229 }
230}
231
232impl std::error::Error for DataError {}
233
234impl From<DataError> for String {
235 fn from(err: DataError) -> String {
236 err.to_string()
237 }
238}
239
240#[derive(Clone, Debug, Serialize, Deserialize)]
245pub struct DataSchema {
246 pub columns: Vec<SchemaColumn>,
247}
248
249#[derive(Clone, Debug, Serialize, Deserialize)]
250pub struct SchemaColumn {
251 pub name: String,
252 pub kind: ColumnKindTag,
253 #[serde(default)]
254 pub levels: Vec<String>,
255}
256
257#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
258#[serde(rename_all = "kebab-case")]
259pub enum ColumnKindTag {
260 Continuous,
261 Binary,
262 Categorical,
263}
264
265#[derive(Clone, Debug, Eq, PartialEq)]
266pub enum UnseenCategoryPolicy {
267 Error,
268 EncodeUnknownForColumns(HashSet<String>),
269}
270
271impl UnseenCategoryPolicy {
272 pub fn encode_unknown_for_columns(columns: HashSet<String>) -> Self {
273 if columns.is_empty() {
274 Self::Error
275 } else {
276 Self::EncodeUnknownForColumns(columns)
277 }
278 }
279
280 fn unseen_code_for(&self, column_name: &str, level_count: usize) -> Option<f64> {
281 match self {
282 Self::Error => None,
283 Self::EncodeUnknownForColumns(columns) => {
284 columns.contains(column_name).then_some(level_count as f64)
285 }
286 }
287 }
288}
289
290#[derive(Clone, Debug)]
291pub struct EncodedDataset {
292 pub headers: Vec<String>,
293 pub values: Array2<f64>,
294 pub schema: DataSchema,
295 pub column_kinds: Vec<ColumnKindTag>,
296}
297
298impl EncodedDataset {
299 pub fn column_map(&self) -> HashMap<String, usize> {
300 self.headers
301 .iter()
302 .enumerate()
303 .map(|(index, header)| (header.clone(), index))
304 .collect()
305 }
306
307 pub fn feature_ranges(&self) -> Vec<(f64, f64)> {
313 self.values
320 .axis_iter(Axis(1))
321 .into_par_iter()
322 .map(|col| {
323 let (lo, hi) =
324 col.iter()
325 .fold((f64::INFINITY, f64::NEG_INFINITY), |(lo, hi), &v| {
326 if v.is_finite() {
327 (lo.min(v), hi.max(v))
328 } else {
329 (lo, hi)
330 }
331 });
332 if !lo.is_finite() || !hi.is_finite() {
333 (0.0, 0.0)
334 } else {
335 (lo, hi)
336 }
337 })
338 .collect()
339 }
340}
341
342fn shared_prefix(a: &str, b: &str) -> usize {
343 a.chars()
344 .zip(b.chars())
345 .take_while(|(ca, cb)| ca == cb)
346 .count()
347}
348
349#[derive(Clone, Copy, Debug, Eq, PartialEq)]
354enum DataFormat {
355 Csv,
356 Tsv,
357 Parquet,
358}
359
360fn detect_format(path: &Path) -> Result<DataFormat, DataError> {
361 let ext = path
362 .extension()
363 .and_then(|s| s.to_str())
364 .unwrap_or_default()
365 .to_ascii_lowercase();
366 match ext.as_str() {
367 "csv" => Ok(DataFormat::Csv),
368 "tsv" | "txt" | "tab" => Ok(DataFormat::Tsv),
369 "parquet" | "pq" | "pqt" => Ok(DataFormat::Parquet),
370 other => Err(DataError::ParseError {
371 reason: format!(
372 "unsupported data file extension '.{other}'; expected csv, tsv, txt, parquet, or pq: '{}'",
373 path.display()
374 ),
375 }),
376 }
377}
378
379pub fn load_dataset_projected(
384 path: &Path,
385 requested_columns: &[String],
386) -> Result<EncodedDataset, DataError> {
387 load_dataset_projected_with_categorical_roles(path, requested_columns, &HashSet::new())
388}
389
390pub fn load_dataset_projected_with_categorical_roles(
412 path: &Path,
413 requested_columns: &[String],
414 categorical_roles: &HashSet<&str>,
415) -> Result<EncodedDataset, DataError> {
416 match detect_format(path)? {
417 DataFormat::Csv => {
418 load_delimited_inferred(path, b',', requested_columns, categorical_roles)
419 }
420 DataFormat::Tsv => {
421 load_delimited_inferred(path, b'\t', requested_columns, categorical_roles)
422 }
423 DataFormat::Parquet => load_parquet_inferred(path, requested_columns, categorical_roles),
424 }
425}
426
427pub fn load_datasetwith_schema_projected(
428 path: &Path,
429 schema: &DataSchema,
430 unseen_policy: UnseenCategoryPolicy,
431 requested_columns: &[String],
432) -> Result<EncodedDataset, DataError> {
433 match detect_format(path)? {
434 DataFormat::Csv => {
435 load_delimited_with_schema(path, b',', schema, unseen_policy, requested_columns)
436 }
437 DataFormat::Tsv => {
438 load_delimited_with_schema(path, b'\t', schema, unseen_policy, requested_columns)
439 }
440 DataFormat::Parquet => {
441 load_parquet_with_schema(path, schema, unseen_policy, requested_columns)
442 }
443 }
444}
445
446pub fn load_csvwith_inferred_schema(path: &Path) -> Result<EncodedDataset, DataError> {
451 load_delimited_inferred(path, b',', &[], &HashSet::new())
452}
453
454pub const CATEGORICAL_CELL_SENTINEL: char = '\u{0}';
469
470pub fn strip_categorical_sentinel(cell: &str) -> (&str, bool) {
473 match cell.strip_prefix(CATEGORICAL_CELL_SENTINEL) {
474 Some(rest) => (rest, true),
475 None => (cell, false),
476 }
477}
478
479fn resolve_requested_columns(
480 all_headers: &[String],
481 requested_columns: &[String],
482) -> Result<Vec<usize>, DataError> {
483 if requested_columns.is_empty() {
484 return Ok((0..all_headers.len()).collect());
485 }
486
487 let requested_set: HashSet<&str> = requested_columns.iter().map(String::as_str).collect();
488 let mut selected = Vec::with_capacity(requested_set.len());
489 for (idx, name) in all_headers.iter().enumerate() {
490 if requested_set.contains(name.as_str()) {
491 selected.push(idx);
492 }
493 }
494
495 if selected.len() != requested_set.len() {
496 let available_map: HashMap<String, usize> = all_headers
497 .iter()
498 .enumerate()
499 .map(|(index, header)| (header.clone(), index))
500 .collect();
501 let missing = requested_columns
502 .iter()
503 .filter(|name| !available_map.contains_key(name.as_str()))
504 .map(|name| {
505 DataError::column_not_found(&available_map, name, Some("requested")).to_string()
506 })
507 .collect::<Vec<_>>();
508 return Err(DataError::SchemaMismatch {
509 reason: missing.join("; "),
510 });
511 }
512
513 Ok(selected)
514}
515
516fn projected_headers(all_headers: &[String], selected_indices: &[usize]) -> Vec<String> {
517 selected_indices
518 .iter()
519 .map(|&idx| all_headers[idx].clone())
520 .collect()
521}
522
523fn load_delimited_inferred(
524 path: &Path,
525 delimiter: u8,
526 requested_columns: &[String],
527 categorical_roles: &HashSet<&str>,
528) -> Result<EncodedDataset, DataError> {
529 let t_open = std::time::Instant::now();
530 let mut rdr = ReaderBuilder::new()
531 .has_headers(true)
532 .delimiter(delimiter)
533 .from_path(path)
534 .map_err(|e| DataError::ParseError {
535 reason: format!("failed to open '{}': {e}", path.display()),
536 })?;
537
538 let all_headers: Vec<String> = rdr
539 .headers()
540 .map_err(|e| DataError::ParseError {
541 reason: format!("failed to read headers: {e}"),
542 })?
543 .iter()
544 .map(|s| s.trim().to_string())
545 .collect();
546 if all_headers.is_empty() {
547 return Err(DataError::EmptyInput {
548 reason: "file has no headers".to_string(),
549 });
550 }
551 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
552 let headers = projected_headers(&all_headers, &selected_indices);
553 let p = headers.len();
554 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
555 if open_ms > 100.0 {
556 log::info!(
557 "[DATA-LOAD] delim_open+headers | n_headers={} | n_proj={} | {:.1}ms",
558 all_headers.len(),
559 p,
560 open_ms
561 );
562 }
563
564 let mut inference = vec![DelimitedInferenceState::default(); p];
568 let mut total_rows: usize = 0;
569 let t_stream = std::time::Instant::now();
570 let mut record = StringRecord::new();
571 while rdr
572 .read_record(&mut record)
573 .map_err(|e| DataError::ParseError {
574 reason: format!("failed reading row: {e}"),
575 })?
576 {
577 if record.len() != all_headers.len() {
578 return Err(DataError::SchemaMismatch {
579 reason: format!(
580 "row width mismatch at row {}: got {} fields, expected {}",
581 total_rows + 1,
582 record.len(),
583 all_headers.len()
584 ),
585 });
586 }
587 total_rows += 1;
588 for (j, &selected_idx) in selected_indices.iter().enumerate() {
589 inference[j].observe(
590 record
591 .get(selected_idx)
592 .expect("record width was checked against the header row above")
593 .trim(),
594 total_rows,
595 &headers[j],
596 )?;
597 }
598 }
599
600 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
601 if stream_ms > 100.0 {
602 log::info!(
603 "[DATA-LOAD] delim_stream | n_rows={} | n_cols={} | {:.1}ms",
604 total_rows,
605 p,
606 stream_ms
607 );
608 }
609
610 if total_rows == 0 {
611 return Err(DataError::EmptyInput {
612 reason: "file has no rows".to_string(),
613 });
614 }
615
616 let t_schema = std::time::Instant::now();
617 let column_kinds = inference
618 .iter()
619 .enumerate()
620 .map(|(j, state)| state.kind(categorical_roles.contains(headers[j].as_str())))
621 .collect::<Vec<_>>();
622 let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
623 if schema_ms > 100.0 {
624 let n_cat = column_kinds
625 .iter()
626 .filter(|k| matches!(k, ColumnKindTag::Categorical))
627 .count();
628 log::info!(
629 "[DATA-LOAD] delim_convert+infer | n_cols={} | n_cat={} | {:.1}ms",
630 p,
631 n_cat,
632 schema_ms
633 );
634 }
635
636 let t_assemble = std::time::Instant::now();
641 let mut values = Array2::<f64>::zeros((total_rows, p));
642 let mut categorical_encoders = (0..p)
643 .map(|j| {
644 matches!(column_kinds[j], ColumnKindTag::Categorical).then(CategoricalEncoder::default)
645 })
646 .collect::<Vec<_>>();
647 let mut encode_rdr = ReaderBuilder::new()
648 .has_headers(true)
649 .delimiter(delimiter)
650 .from_path(path)
651 .map_err(|e| DataError::ParseError {
652 reason: format!("failed to reopen '{}': {e}", path.display()),
653 })?;
654 encode_rdr.headers().map_err(|e| DataError::ParseError {
655 reason: format!("failed to reread headers: {e}"),
656 })?;
657 let mut encoded_rows = 0usize;
658 while encode_rdr
659 .read_record(&mut record)
660 .map_err(|e| DataError::ParseError {
661 reason: format!("failed reading row: {e}"),
662 })?
663 {
664 if record.len() != all_headers.len() {
665 return Err(DataError::SchemaMismatch {
666 reason: format!(
667 "row width mismatch at row {}: got {} fields, expected {}",
668 encoded_rows + 1,
669 record.len(),
670 all_headers.len()
671 ),
672 });
673 }
674 if encoded_rows >= total_rows {
675 return Err(DataError::SchemaMismatch {
676 reason: "data file changed while its schema was being discovered".to_string(),
677 });
678 }
679 for (j, &selected_idx) in selected_indices.iter().enumerate() {
680 let raw = record
681 .get(selected_idx)
682 .expect("record width was checked against the header row above")
683 .trim();
684 values[[encoded_rows, j]] = match column_kinds[j] {
685 ColumnKindTag::Continuous | ColumnKindTag::Binary => {
686 parse_inferred_numeric_cell(raw, encoded_rows + 1, &headers[j])?
687 }
688 ColumnKindTag::Categorical => {
689 if raw.is_empty() {
690 return Err(DataError::EmptyInput {
691 reason: format!(
692 "empty field at row {}, column '{}'",
693 encoded_rows + 1,
694 &headers[j]
695 ),
696 });
697 }
698 categorical_encoders[j]
699 .as_mut()
700 .expect("categorical encoder")
701 .encode(raw) as f64
702 }
703 };
704 }
705 encoded_rows += 1;
706 }
707 if encoded_rows != total_rows {
708 return Err(DataError::SchemaMismatch {
709 reason: "data file changed while its schema was being discovered".to_string(),
710 });
711 }
712
713 let mut levels = vec![Vec::<String>::new(); p];
714 for (j, encoder) in categorical_encoders.into_iter().enumerate() {
715 if let Some(encoder) = encoder {
716 levels[j] = encoder.finish(values.column_mut(j), LevelOrder::Canonical);
717 }
718 }
719 let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
720 if assemble_ms > 100.0 {
721 log::info!(
722 "[DATA-LOAD] delim_assemble_array2 | n_rows={} | n_cols={} | {:.1}ms",
723 total_rows,
724 p,
725 assemble_ms
726 );
727 }
728
729 let schema = DataSchema {
730 columns: headers
731 .iter()
732 .enumerate()
733 .map(|(j, name)| SchemaColumn {
734 name: name.clone(),
735 kind: column_kinds[j],
736 levels: std::mem::take(&mut levels[j]),
737 })
738 .collect(),
739 };
740 Ok(EncodedDataset {
741 headers,
742 values,
743 schema,
744 column_kinds,
745 })
746}
747
748#[derive(Clone, Copy)]
749struct DelimitedInferenceState {
750 all_numeric: bool,
751 all_binary: bool,
752 saw_numeric: bool,
756}
757
758impl Default for DelimitedInferenceState {
759 fn default() -> Self {
760 Self {
761 all_numeric: true,
762 all_binary: true,
763 saw_numeric: false,
764 }
765 }
766}
767
768impl DelimitedInferenceState {
769 fn observe(&mut self, raw: &str, row: usize, header: &str) -> Result<(), DataError> {
770 if raw.is_empty() {
771 return Err(DataError::EmptyInput {
772 reason: format!("empty field at row {row}, column '{header}'"),
773 });
774 }
775 if is_missing_marker(raw) {
779 return Ok(());
780 }
781 match raw.parse::<f64>() {
782 Ok(value) => {
783 self.saw_numeric = true;
784 if !value.is_finite() {
785 return Err(DataError::InvalidValue {
786 reason: format!("non-finite value at row {row}, column '{header}'"),
787 });
788 }
789 if (value - 0.0).abs() >= 1e-12 && (value - 1.0).abs() >= 1e-12 {
790 self.all_binary = false;
791 }
792 }
793 Err(_) => {
794 self.all_numeric = false;
795 self.all_binary = false;
796 }
797 }
798 Ok(())
799 }
800
801 fn kind(self, force_categorical: bool) -> ColumnKindTag {
802 if force_categorical || !(self.all_numeric && self.saw_numeric) {
803 ColumnKindTag::Categorical
804 } else if self.all_binary {
805 ColumnKindTag::Binary
806 } else {
807 ColumnKindTag::Continuous
808 }
809 }
810}
811
812fn is_missing_marker(raw: &str) -> bool {
829 matches!(
830 raw.trim().to_ascii_uppercase().as_str(),
831 "NA" | "N/A" | "NULL"
832 )
833}
834
835fn parse_inferred_numeric_cell(raw: &str, row: usize, header: &str) -> Result<f64, DataError> {
836 if raw.is_empty() {
837 return Err(DataError::EmptyInput {
838 reason: format!("empty field at row {row}, column '{header}'"),
839 });
840 }
841 if is_missing_marker(raw) {
846 return Ok(f64::NAN);
847 }
848 let value = raw
849 .parse::<f64>()
850 .map_err(|error| DataError::EncodingFailure {
851 reason: format!(
852 "failed to parse numeric value '{raw}' at row {row}, column '{header}': {error}"
853 ),
854 })?;
855 if !value.is_finite() {
856 return Err(DataError::InvalidValue {
857 reason: format!("non-finite value at row {row}, column '{header}'"),
858 });
859 }
860 Ok(value)
861}
862
863#[derive(Clone, Copy)]
864enum LevelOrder {
865 Encounter,
866 Canonical,
867}
868
869#[derive(Default)]
878struct CategoricalEncoder {
879 encounter_codes: HashMap<String, usize>,
880}
881
882impl CategoricalEncoder {
883 fn encode(&mut self, label: &str) -> usize {
884 if let Some(&code) = self.encounter_codes.get(label) {
885 return code;
886 }
887 let code = self.encounter_codes.len();
888 self.encounter_codes.insert(label.to_owned(), code);
889 code
890 }
891
892 fn finish(self, mut encoded: ArrayViewMut1<'_, f64>, order: LevelOrder) -> Vec<String> {
893 match order {
894 LevelOrder::Encounter => {
895 let mut levels = std::iter::repeat_with(|| None)
896 .take(self.encounter_codes.len())
897 .collect::<Vec<Option<String>>>();
898 for (level, old_code) in self.encounter_codes {
899 levels[old_code] = Some(level);
900 }
901 levels
902 .into_iter()
903 .map(|level| level.expect("encounter code must name one level"))
904 .collect()
905 }
906 LevelOrder::Canonical => {
907 let mut levels_with_old_codes =
908 self.encounter_codes.into_iter().collect::<Vec<_>>();
909 levels_with_old_codes
910 .sort_by(|(a, _), (b, _)| natural_level_cmp(a.as_str(), b.as_str()));
911 let mut remap = vec![0usize; levels_with_old_codes.len()];
912 for (new_code, (_, old_code)) in levels_with_old_codes.iter().enumerate() {
913 remap[*old_code] = new_code;
914 }
915 for code in encoded.iter_mut() {
916 *code = remap[*code as usize] as f64;
917 }
918 levels_with_old_codes
919 .into_iter()
920 .map(|(level, _)| level)
921 .collect()
922 }
923 }
924 }
925}
926
927fn load_delimited_with_schema(
928 path: &Path,
929 delimiter: u8,
930 schema: &DataSchema,
931 unseen_policy: UnseenCategoryPolicy,
932 requested_columns: &[String],
933) -> Result<EncodedDataset, DataError> {
934 let t_open = std::time::Instant::now();
935 let mut rdr = ReaderBuilder::new()
936 .has_headers(true)
937 .delimiter(delimiter)
938 .from_path(path)
939 .map_err(|e| DataError::ParseError {
940 reason: format!("failed to open '{}': {e}", path.display()),
941 })?;
942
943 let all_headers: Vec<String> = rdr
944 .headers()
945 .map_err(|e| DataError::ParseError {
946 reason: format!("failed to read headers: {e}"),
947 })?
948 .iter()
949 .map(|s| s.trim().to_string())
950 .collect();
951 if all_headers.is_empty() {
952 return Err(DataError::EmptyInput {
953 reason: "file has no headers".to_string(),
954 });
955 }
956 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
957 let headers = projected_headers(&all_headers, &selected_indices);
958 let p = headers.len();
959 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
960 if open_ms > 100.0 {
961 log::info!(
962 "[DATA-LOAD] delim_schema_open+headers | n_headers={} | n_proj={} | {:.1}ms",
963 all_headers.len(),
964 p,
965 open_ms
966 );
967 }
968
969 let schema_byname: HashMap<&str, &SchemaColumn> = schema
971 .columns
972 .iter()
973 .map(|c| (c.name.as_str(), c))
974 .collect();
975
976 let mut col_meta = Vec::<ColMeta>::with_capacity(p);
977 for name in &headers {
978 if let Some(sc) = schema_byname.get(name.as_str()) {
979 let level_map = if matches!(sc.kind, ColumnKindTag::Categorical) {
980 Some(
981 sc.levels
982 .iter()
983 .enumerate()
984 .map(|(idx, v)| (v.as_str(), idx as f64))
985 .collect::<HashMap<_, _>>(),
986 )
987 } else {
988 None
989 };
990 col_meta.push(ColMeta {
991 kind: sc.kind,
992 level_map,
993 schema_col: (*sc).clone(),
994 });
995 } else {
996 col_meta.push(ColMeta {
998 kind: ColumnKindTag::Continuous, level_map: None,
1000 schema_col: SchemaColumn {
1001 name: name.clone(),
1002 kind: ColumnKindTag::Continuous,
1003 levels: Vec::new(),
1004 },
1005 });
1006 }
1007 }
1008
1009 let needs_inference: Vec<bool> = headers
1011 .iter()
1012 .map(|h| !schema_byname.contains_key(h.as_str()))
1013 .collect();
1014
1015 if needs_inference.iter().all(|needs| !needs) {
1021 let t_stream = std::time::Instant::now();
1022 let mut flat_values = Vec::<f64>::new();
1023 let mut total_rows = 0usize;
1024 let mut record = StringRecord::new();
1025 while rdr
1026 .read_record(&mut record)
1027 .map_err(|e| DataError::ParseError {
1028 reason: format!("failed reading row: {e}"),
1029 })?
1030 {
1031 if record.len() != all_headers.len() {
1032 return Err(DataError::SchemaMismatch {
1033 reason: format!(
1034 "row width mismatch at row {}: got {} fields, expected {}",
1035 total_rows + 1,
1036 record.len(),
1037 all_headers.len()
1038 ),
1039 });
1040 }
1041 total_rows += 1;
1042 for j in 0..p {
1043 let raw = record
1044 .get(selected_indices[j])
1045 .expect("record width was checked against the header row above")
1046 .trim();
1047 flat_values.push(parse_cell_with_schema(
1048 raw,
1049 &col_meta[j],
1050 total_rows,
1051 &headers[j],
1052 &unseen_policy,
1053 )?);
1054 }
1055 }
1056 if total_rows == 0 {
1057 return Err(DataError::EmptyInput {
1058 reason: "file has no rows".to_string(),
1059 });
1060 }
1061 let values = Array2::from_shape_vec((total_rows, p), flat_values).map_err(|error| {
1062 DataError::EncodingFailure {
1063 reason: format!("failed to assemble schema-guided delimited matrix: {error}"),
1064 }
1065 })?;
1066 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
1067 if stream_ms > 100.0 {
1068 log::info!(
1069 "[DATA-LOAD] delim_schema_direct | n_rows={} | n_cols={} | {:.1}ms",
1070 total_rows,
1071 p,
1072 stream_ms
1073 );
1074 }
1075 let column_kinds = col_meta.iter().map(|meta| meta.kind).collect();
1076 let schema_out = DataSchema {
1077 columns: col_meta.into_iter().map(|meta| meta.schema_col).collect(),
1078 };
1079 return Ok(EncodedDataset {
1080 headers,
1081 values,
1082 schema: schema_out,
1083 column_kinds,
1084 });
1085 }
1086
1087 let mut inference = vec![DelimitedInferenceState::default(); p];
1090 let mut total_rows: usize = 0;
1091 let t_stream = std::time::Instant::now();
1092 let mut record = StringRecord::new();
1093 while rdr
1094 .read_record(&mut record)
1095 .map_err(|e| DataError::ParseError {
1096 reason: format!("failed reading row: {e}"),
1097 })?
1098 {
1099 if record.len() != all_headers.len() {
1100 return Err(DataError::SchemaMismatch {
1101 reason: format!(
1102 "row width mismatch at row {}: got {} fields, expected {}",
1103 total_rows + 1,
1104 record.len(),
1105 all_headers.len()
1106 ),
1107 });
1108 }
1109 total_rows += 1;
1110
1111 for j in 0..p {
1112 let raw = record
1113 .get(selected_indices[j])
1114 .expect("record width was checked against the header row above")
1115 .trim();
1116 if needs_inference[j] {
1117 inference[j].observe(raw, total_rows, &headers[j])?;
1118 } else {
1119 parse_cell_with_schema(raw, &col_meta[j], total_rows, &headers[j], &unseen_policy)?;
1120 }
1121 }
1122 }
1123
1124 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
1125 if stream_ms > 100.0 {
1126 let n_inf = needs_inference.iter().filter(|x| **x).count();
1127 log::info!(
1128 "[DATA-LOAD] delim_schema_stream | n_rows={} | n_cols={} | n_inf={} | {:.1}ms",
1129 total_rows,
1130 p,
1131 n_inf,
1132 stream_ms
1133 );
1134 }
1135
1136 if total_rows == 0 {
1137 return Err(DataError::EmptyInput {
1138 reason: "file has no rows".to_string(),
1139 });
1140 }
1141
1142 let t_finalize = std::time::Instant::now();
1143 for j in 0..p {
1144 if needs_inference[j] {
1145 let kind = inference[j].kind(false);
1146 col_meta[j].kind = kind;
1147 col_meta[j].schema_col.kind = kind;
1148 }
1149 }
1150 let finalize_ms = t_finalize.elapsed().as_secs_f64() * 1000.0;
1151 if finalize_ms > 100.0 {
1152 log::info!(
1153 "[DATA-LOAD] delim_schema_finalize | n_cols={} | {:.1}ms",
1154 p,
1155 finalize_ms
1156 );
1157 }
1158
1159 let t_assemble = std::time::Instant::now();
1163 let mut values = Array2::<f64>::zeros((total_rows, p));
1164 let mut inferred_encoders = (0..p)
1165 .map(|j| {
1166 (needs_inference[j] && matches!(col_meta[j].kind, ColumnKindTag::Categorical))
1167 .then(CategoricalEncoder::default)
1168 })
1169 .collect::<Vec<_>>();
1170 let mut encode_rdr = ReaderBuilder::new()
1171 .has_headers(true)
1172 .delimiter(delimiter)
1173 .from_path(path)
1174 .map_err(|e| DataError::ParseError {
1175 reason: format!("failed to reopen '{}': {e}", path.display()),
1176 })?;
1177 encode_rdr.headers().map_err(|e| DataError::ParseError {
1178 reason: format!("failed to reread headers: {e}"),
1179 })?;
1180 let mut encoded_rows = 0usize;
1181 while encode_rdr
1182 .read_record(&mut record)
1183 .map_err(|e| DataError::ParseError {
1184 reason: format!("failed reading row: {e}"),
1185 })?
1186 {
1187 if record.len() != all_headers.len() {
1188 return Err(DataError::SchemaMismatch {
1189 reason: format!(
1190 "row width mismatch at row {}: got {} fields, expected {}",
1191 encoded_rows + 1,
1192 record.len(),
1193 all_headers.len()
1194 ),
1195 });
1196 }
1197 if encoded_rows >= total_rows {
1198 return Err(DataError::SchemaMismatch {
1199 reason: "data file changed while its schema was being discovered".to_string(),
1200 });
1201 }
1202 for j in 0..p {
1203 let raw = record
1204 .get(selected_indices[j])
1205 .expect("record width was checked against the header row above")
1206 .trim();
1207 values[[encoded_rows, j]] = if !needs_inference[j] {
1208 parse_cell_with_schema(
1209 raw,
1210 &col_meta[j],
1211 encoded_rows + 1,
1212 &headers[j],
1213 &unseen_policy,
1214 )?
1215 } else {
1216 match col_meta[j].kind {
1217 ColumnKindTag::Continuous | ColumnKindTag::Binary => {
1218 parse_inferred_numeric_cell(raw, encoded_rows + 1, &headers[j])?
1219 }
1220 ColumnKindTag::Categorical => {
1221 if raw.is_empty() {
1222 return Err(DataError::EmptyInput {
1223 reason: format!(
1224 "empty field at row {}, column '{}'",
1225 encoded_rows + 1,
1226 &headers[j]
1227 ),
1228 });
1229 }
1230 let encoder = inferred_encoders[j]
1231 .as_mut()
1232 .expect("inferred categorical encoder");
1233 encoder.encode(raw) as f64
1234 }
1235 }
1236 };
1237 }
1238 encoded_rows += 1;
1239 }
1240 if encoded_rows != total_rows {
1241 return Err(DataError::SchemaMismatch {
1242 reason: "data file changed while its schema was being discovered".to_string(),
1243 });
1244 }
1245 for (j, encoder) in inferred_encoders.into_iter().enumerate() {
1246 if let Some(encoder) = encoder {
1247 col_meta[j].schema_col.levels =
1248 encoder.finish(values.column_mut(j), LevelOrder::Canonical);
1249 }
1250 }
1251 let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
1252 if assemble_ms > 100.0 {
1253 log::info!(
1254 "[DATA-LOAD] delim_schema_assemble | n_rows={} | n_cols={} | {:.1}ms",
1255 total_rows,
1256 p,
1257 assemble_ms
1258 );
1259 }
1260
1261 let column_kinds = col_meta.iter().map(|meta| meta.kind).collect();
1262 let schema_out = DataSchema {
1263 columns: col_meta.into_iter().map(|m| m.schema_col).collect(),
1264 };
1265 Ok(EncodedDataset {
1266 headers,
1267 values,
1268 schema: schema_out,
1269 column_kinds,
1270 })
1271}
1272
1273fn parse_cell_with_schema(
1274 raw: &str,
1275 meta: &ColMeta<'_>,
1276 row: usize,
1277 col_name: &str,
1278 unseen_policy: &UnseenCategoryPolicy,
1279) -> Result<f64, DataError> {
1280 let val = match meta.kind {
1281 ColumnKindTag::Continuous if is_missing_marker(raw) => f64::NAN,
1284 ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
1285 DataError::SchemaMismatch {
1286 reason: format!(
1287 "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
1288 col_name, row, raw, err
1289 ),
1290 }
1291 })?,
1292 ColumnKindTag::Binary if is_missing_marker(raw) => f64::NAN,
1293 ColumnKindTag::Binary => {
1294 let v = raw
1295 .parse::<f64>()
1296 .map_err(|err| DataError::SchemaMismatch {
1297 reason: format!(
1298 "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
1299 col_name, row, raw, err
1300 ),
1301 })?;
1302 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
1303 return Err(DataError::SchemaMismatch {
1304 reason: format!(
1305 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
1306 col_name, row, v
1307 ),
1308 });
1309 }
1310 v
1311 }
1312 ColumnKindTag::Categorical => {
1313 let map = meta
1314 .level_map
1315 .as_ref()
1316 .ok_or_else(|| DataError::EncodingFailure {
1317 reason: "internal categorical schema map missing".to_string(),
1318 })?;
1319 match map.get(raw) {
1320 Some(v) => *v,
1321 None => unseen_policy
1322 .unseen_code_for(col_name, meta.schema_col.levels.len())
1323 .ok_or_else(|| DataError::SchemaMismatch {
1324 reason: format!(
1325 "unseen level '{}' in categorical column '{}' at row {}",
1326 raw, col_name, row
1327 ),
1328 })?,
1329 }
1330 }
1331 };
1332 if !val.is_finite() && !is_missing_marker(raw) {
1336 return Err(DataError::InvalidValue {
1337 reason: format!("non-finite value at row {}, column '{}'", row, col_name),
1338 });
1339 }
1340 Ok(val)
1341}
1342
1343struct ColMeta<'a> {
1346 kind: ColumnKindTag,
1347 level_map: Option<HashMap<&'a str, f64>>,
1348 schema_col: SchemaColumn,
1349}
1350
1351fn arrow_field_is_string(dt: &arrow::datatypes::DataType) -> bool {
1364 use arrow::datatypes::DataType;
1365 match dt {
1366 DataType::Utf8 | DataType::LargeUtf8 => true,
1367 DataType::Dictionary(_, value_type) => arrow_field_is_string(value_type),
1368 _ => false,
1369 }
1370}
1371
1372fn write_arrow_numeric_values(
1373 values: impl IntoIterator<Item = f64>,
1374 base_row: usize,
1375 header: &str,
1376 mut output: ArrayViewMut1<'_, f64>,
1377 categorical_encoder: Option<&mut CategoricalEncoder>,
1378 all_binary: &mut bool,
1379) -> Result<(), DataError> {
1380 match categorical_encoder {
1381 Some(encoder) => {
1382 for (batch_row, value) in values.into_iter().enumerate() {
1383 if !value.is_finite() {
1384 return Err(DataError::InvalidValue {
1385 reason: format!(
1386 "non-finite value at row {}, column '{}'",
1387 base_row + batch_row + 1,
1388 header
1389 ),
1390 });
1391 }
1392 output[batch_row] = encoder.encode(&value.to_string()) as f64;
1393 }
1394 }
1395 None => {
1396 for (batch_row, value) in values.into_iter().enumerate() {
1397 if !value.is_finite() {
1398 return Err(DataError::InvalidValue {
1399 reason: format!(
1400 "non-finite value at row {}, column '{}'",
1401 base_row + batch_row + 1,
1402 header
1403 ),
1404 });
1405 }
1406 if (value - 0.0).abs() >= 1e-12 && (value - 1.0).abs() >= 1e-12 {
1407 *all_binary = false;
1408 }
1409 output[batch_row] = value;
1410 }
1411 }
1412 }
1413 Ok(())
1414}
1415
1416fn reject_arrow_null_values(
1417 col: &dyn arrow::array::Array,
1418 base_row: usize,
1419 header: &str,
1420) -> Result<(), DataError> {
1421 let Some(nulls) = col.logical_nulls() else {
1422 return Ok(());
1423 };
1424 if let Some(batch_row) = (0..col.len()).find(|&row| nulls.is_null(row)) {
1425 return Err(DataError::InvalidValue {
1426 reason: format!(
1427 "null value at row {}, column '{}'",
1428 base_row + batch_row + 1,
1429 header
1430 ),
1431 });
1432 }
1433 Ok(())
1434}
1435
1436fn arrow_dictionary_string_value_at<'a, K>(
1437 col: &'a dyn arrow::array::Array,
1438 index: usize,
1439 logical_row: usize,
1440 header: &str,
1441) -> Result<&'a str, DataError>
1442where
1443 K: arrow::datatypes::ArrowDictionaryKeyType,
1444{
1445 use arrow::array::DictionaryArray;
1446
1447 let dictionary = col
1448 .as_any()
1449 .downcast_ref::<DictionaryArray<K>>()
1450 .ok_or_else(|| DataError::EncodingFailure {
1451 reason: format!(
1452 "Arrow dictionary column '{}' did not match its declared key type",
1453 header
1454 ),
1455 })?;
1456 let value_index = dictionary
1457 .key(index)
1458 .ok_or_else(|| DataError::InvalidValue {
1459 reason: format!("null value at row {logical_row}, column '{header}'"),
1460 })?;
1461 if value_index >= dictionary.values().len() {
1462 return Err(DataError::EncodingFailure {
1463 reason: format!(
1464 "Arrow dictionary column '{}' has out-of-range key {} at row {}",
1465 header, value_index, logical_row
1466 ),
1467 });
1468 }
1469 arrow_string_value_at(
1470 dictionary.values().as_ref(),
1471 value_index,
1472 logical_row,
1473 header,
1474 )
1475}
1476
1477fn arrow_string_value_at<'a>(
1480 col: &'a dyn arrow::array::Array,
1481 index: usize,
1482 logical_row: usize,
1483 header: &str,
1484) -> Result<&'a str, DataError> {
1485 use arrow::array::{LargeStringArray, StringArray};
1486 use arrow::datatypes::{
1487 DataType, Int8Type, Int16Type, Int32Type, Int64Type, UInt8Type, UInt16Type, UInt32Type,
1488 UInt64Type,
1489 };
1490
1491 if index >= col.len() {
1492 return Err(DataError::EncodingFailure {
1493 reason: format!(
1494 "Arrow string column '{}' has out-of-range index {} at row {}",
1495 header, index, logical_row
1496 ),
1497 });
1498 }
1499 if col.is_null(index) {
1500 return Err(DataError::InvalidValue {
1501 reason: format!("null value at row {logical_row}, column '{header}'"),
1502 });
1503 }
1504
1505 match col.data_type() {
1506 DataType::Utf8 => col
1507 .as_any()
1508 .downcast_ref::<StringArray>()
1509 .map(|array| array.value(index))
1510 .ok_or_else(|| DataError::EncodingFailure {
1511 reason: format!("Arrow column '{}' could not be read as Utf8", header),
1512 }),
1513 DataType::LargeUtf8 => col
1514 .as_any()
1515 .downcast_ref::<LargeStringArray>()
1516 .map(|array| array.value(index))
1517 .ok_or_else(|| DataError::EncodingFailure {
1518 reason: format!("Arrow column '{}' could not be read as LargeUtf8", header),
1519 }),
1520 DataType::Dictionary(key_type, _) => match key_type.as_ref() {
1521 DataType::Int8 => {
1522 arrow_dictionary_string_value_at::<Int8Type>(col, index, logical_row, header)
1523 }
1524 DataType::Int16 => {
1525 arrow_dictionary_string_value_at::<Int16Type>(col, index, logical_row, header)
1526 }
1527 DataType::Int32 => {
1528 arrow_dictionary_string_value_at::<Int32Type>(col, index, logical_row, header)
1529 }
1530 DataType::Int64 => {
1531 arrow_dictionary_string_value_at::<Int64Type>(col, index, logical_row, header)
1532 }
1533 DataType::UInt8 => {
1534 arrow_dictionary_string_value_at::<UInt8Type>(col, index, logical_row, header)
1535 }
1536 DataType::UInt16 => {
1537 arrow_dictionary_string_value_at::<UInt16Type>(col, index, logical_row, header)
1538 }
1539 DataType::UInt32 => {
1540 arrow_dictionary_string_value_at::<UInt32Type>(col, index, logical_row, header)
1541 }
1542 DataType::UInt64 => {
1543 arrow_dictionary_string_value_at::<UInt64Type>(col, index, logical_row, header)
1544 }
1545 other => Err(DataError::InvalidValue {
1546 reason: format!(
1547 "unsupported Arrow dictionary key type {:?} for column '{}'",
1548 other, header
1549 ),
1550 }),
1551 },
1552 other => Err(DataError::InvalidValue {
1553 reason: format!(
1554 "unsupported Arrow string column type {:?} for column '{}'",
1555 other, header
1556 ),
1557 }),
1558 }
1559}
1560
1561fn decode_arrow_batch_column_into(
1568 col: &dyn arrow::array::Array,
1569 base_row: usize,
1570 header: &str,
1571 is_string_col: bool,
1572 mut output: ArrayViewMut1<'_, f64>,
1573 mut categorical_encoder: Option<&mut CategoricalEncoder>,
1574 all_binary: &mut bool,
1575) -> Result<(), DataError> {
1576 use arrow::array::{
1577 BooleanArray, Float32Array, Float64Array, Int8Array, Int16Array, Int32Array, Int64Array,
1578 UInt8Array, UInt16Array, UInt32Array, UInt64Array,
1579 };
1580 use arrow::datatypes::DataType;
1581
1582 let n_rows = output.len();
1583 if col.len() != n_rows {
1584 return Err(DataError::SchemaMismatch {
1585 reason: format!(
1586 "Arrow column '{}' has {} rows, but its record batch has {}",
1587 header,
1588 col.len(),
1589 n_rows
1590 ),
1591 });
1592 }
1593 reject_arrow_null_values(col, base_row, header)?;
1594
1595 if is_string_col {
1596 let encoder =
1597 categorical_encoder
1598 .as_deref_mut()
1599 .ok_or_else(|| DataError::EncodingFailure {
1600 reason: format!("categorical Arrow encoder missing for column '{header}'"),
1601 })?;
1602 for batch_row in 0..n_rows {
1603 let label = arrow_string_value_at(col, batch_row, base_row + batch_row + 1, header)?;
1604 output[batch_row] = encoder.encode(label) as f64;
1605 }
1606 return Ok(());
1607 }
1608
1609 let decoded_col;
1616 let col: &dyn arrow::array::Array = if let DataType::Dictionary(_, value_type) = col.data_type()
1617 {
1618 decoded_col = arrow::compute::cast(col, value_type).map_err(|e| DataError::ParseError {
1619 reason: format!(
1620 "failed to decode dictionary-encoded numeric column '{}': {e}",
1621 header
1622 ),
1623 })?;
1624 decoded_col.as_ref()
1625 } else {
1626 col
1627 };
1628 reject_arrow_null_values(col, base_row, header)?;
1631
1632 macro_rules! write_primitive {
1633 ($array_type:ty, $convert:expr) => {{
1634 let array = col
1635 .as_any()
1636 .downcast_ref::<$array_type>()
1637 .expect("array type is the one this `col.data_type()` arm matched");
1638 write_arrow_numeric_values(
1639 array.values().iter().copied().map($convert),
1640 base_row,
1641 header,
1642 output,
1643 categorical_encoder,
1644 all_binary,
1645 )
1646 }};
1647 }
1648
1649 match col.data_type() {
1650 DataType::Float64 => write_primitive!(Float64Array, |value: f64| value),
1651 DataType::Float32 => write_primitive!(Float32Array, |value: f32| value as f64),
1652 DataType::Int64 => write_primitive!(Int64Array, |value: i64| value as f64),
1653 DataType::Int32 => write_primitive!(Int32Array, |value: i32| value as f64),
1654 DataType::Int16 => write_primitive!(Int16Array, |value: i16| value as f64),
1655 DataType::Int8 => write_primitive!(Int8Array, |value: i8| value as f64),
1656 DataType::UInt64 => write_primitive!(UInt64Array, |value: u64| value as f64),
1657 DataType::UInt32 => write_primitive!(UInt32Array, |value: u32| value as f64),
1658 DataType::UInt16 => write_primitive!(UInt16Array, |value: u16| value as f64),
1659 DataType::UInt8 => write_primitive!(UInt8Array, |value: u8| value as f64),
1660 DataType::Boolean => {
1661 let arr = col
1662 .as_any()
1663 .downcast_ref::<BooleanArray>()
1664 .expect("array type is BooleanArray in the DataType::Boolean arm");
1665 write_arrow_numeric_values(
1666 (0..n_rows).map(|i| if arr.value(i) { 1.0 } else { 0.0 }),
1667 base_row,
1668 header,
1669 output,
1670 categorical_encoder,
1671 all_binary,
1672 )
1673 }
1674 other => Err(DataError::InvalidValue {
1675 reason: format!(
1676 "unsupported Arrow column type {:?} for column '{}'",
1677 other, header
1678 ),
1679 }),
1680 }
1681}
1682
1683pub fn encode_arrow_record_batch_reader_with_inferred_schema(
1697 reader: &mut dyn arrow::record_batch::RecordBatchReader,
1698 headers: Vec<String>,
1699) -> Result<EncodedDataset, DataError> {
1700 if headers.is_empty() {
1701 return Err(DataError::EmptyInput {
1702 reason: "Arrow table must have at least one header column".to_string(),
1703 });
1704 }
1705
1706 let mut seen_headers = HashSet::<&str>::with_capacity(headers.len());
1707 for (column, header) in headers.iter().enumerate() {
1708 if header.trim().is_empty() {
1709 return Err(DataError::EmptyInput {
1710 reason: format!("Arrow header at column {} cannot be empty", column + 1),
1711 });
1712 }
1713 if !seen_headers.insert(header.as_str()) {
1714 return Err(DataError::SchemaMismatch {
1715 reason: format!("duplicate Arrow header '{}'", header),
1716 });
1717 }
1718 }
1719
1720 let arrow_schema = reader.schema();
1721 let p = headers.len();
1722 if arrow_schema.fields().len() != p {
1723 return Err(DataError::SchemaMismatch {
1724 reason: format!(
1725 "Arrow schema has {} columns, but {} normalized headers were supplied",
1726 arrow_schema.fields().len(),
1727 p
1728 ),
1729 });
1730 }
1731
1732 let is_string_col = arrow_schema
1733 .fields()
1734 .iter()
1735 .map(|field| arrow_field_is_string(field.data_type()))
1736 .collect::<Vec<_>>();
1737 let mut all_binary = vec![true; p];
1738 let mut categorical_encoders = is_string_col
1739 .iter()
1740 .map(|&is_string| is_string.then(CategoricalEncoder::default))
1741 .collect::<Vec<_>>();
1742 let mut encoded_values = Vec::<f64>::new();
1743 let mut rows_seen = 0usize;
1744
1745 for batch_result in reader {
1746 let batch = batch_result.map_err(|error| DataError::ParseError {
1747 reason: format!("failed to read Arrow record batch: {error}"),
1748 })?;
1749 if batch.num_columns() != p {
1750 return Err(DataError::SchemaMismatch {
1751 reason: format!(
1752 "Arrow record batch has {} columns, but {} normalized headers were supplied",
1753 batch.num_columns(),
1754 p
1755 ),
1756 });
1757 }
1758 for j in 0..p {
1759 let expected = arrow_schema.field(j).data_type();
1760 let actual = batch.column(j).data_type();
1761 if actual != expected {
1762 return Err(DataError::SchemaMismatch {
1763 reason: format!(
1764 "Arrow column '{}' changed type between schema and batch: expected {:?}, got {:?}",
1765 headers[j], expected, actual
1766 ),
1767 });
1768 }
1769 }
1770
1771 let n_rows = batch.num_rows();
1772 let batch_values = n_rows
1773 .checked_mul(p)
1774 .ok_or_else(|| DataError::EncodingFailure {
1775 reason: "Arrow batch dimensions do not fit in memory address space".to_string(),
1776 })?;
1777 let next_len = encoded_values
1778 .len()
1779 .checked_add(batch_values)
1780 .ok_or_else(|| DataError::EncodingFailure {
1781 reason: "Arrow dataset dimensions do not fit in memory address space".to_string(),
1782 })?;
1783 encoded_values
1784 .try_reserve(batch_values)
1785 .map_err(|error| DataError::EncodingFailure {
1786 reason: format!("failed to reserve Arrow dataset storage: {error}"),
1787 })?;
1788 let batch_offset = encoded_values.len();
1789 encoded_values.resize(next_len, 0.0);
1790 let mut batch_output = ndarray::ArrayViewMut2::from_shape(
1791 (n_rows, p),
1792 &mut encoded_values[batch_offset..next_len],
1793 )
1794 .map_err(|error| DataError::EncodingFailure {
1795 reason: format!("failed to shape Arrow batch output: {error}"),
1796 })?;
1797
1798 let decoded_columns = batch_output
1799 .axis_iter_mut(Axis(1))
1800 .into_par_iter()
1801 .zip(categorical_encoders.par_iter_mut())
1802 .zip(all_binary.par_iter_mut())
1803 .enumerate()
1804 .map(|(j, ((output, encoder), column_all_binary))| {
1805 decode_arrow_batch_column_into(
1806 batch.column(j).as_ref(),
1807 rows_seen,
1808 &headers[j],
1809 is_string_col[j],
1810 output,
1811 encoder.as_mut(),
1812 column_all_binary,
1813 )
1814 })
1815 .collect::<Vec<_>>();
1816 for decoded in decoded_columns {
1817 decoded?;
1818 }
1819 rows_seen = rows_seen
1820 .checked_add(n_rows)
1821 .ok_or_else(|| DataError::EncodingFailure {
1822 reason: "Arrow row count does not fit in memory address space".to_string(),
1823 })?;
1824 }
1825
1826 if rows_seen == 0 {
1827 return Err(DataError::EmptyInput {
1828 reason: "Arrow table data cannot be empty".to_string(),
1829 });
1830 }
1831
1832 let mut values = Array2::from_shape_vec((rows_seen, p), encoded_values).map_err(|error| {
1833 DataError::EncodingFailure {
1834 reason: format!("failed to shape encoded Arrow dataset: {error}"),
1835 }
1836 })?;
1837 let mut levels = vec![Vec::<String>::new(); p];
1838 for (j, encoder) in categorical_encoders.into_iter().enumerate() {
1839 if let Some(encoder) = encoder {
1840 levels[j] = encoder.finish(values.column_mut(j), LevelOrder::Canonical);
1841 }
1842 }
1843
1844 let mut schema_columns = Vec::<SchemaColumn>::with_capacity(p);
1845 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
1846 for (j, name) in headers.iter().enumerate() {
1847 let kind = if is_string_col[j] {
1848 ColumnKindTag::Categorical
1849 } else if all_binary[j] {
1850 ColumnKindTag::Binary
1851 } else {
1852 ColumnKindTag::Continuous
1853 };
1854 column_kinds.push(kind);
1855 schema_columns.push(SchemaColumn {
1856 name: name.clone(),
1857 kind,
1858 levels: std::mem::take(&mut levels[j]),
1859 });
1860 }
1861
1862 Ok(EncodedDataset {
1863 headers,
1864 values,
1865 schema: DataSchema {
1866 columns: schema_columns,
1867 },
1868 column_kinds,
1869 })
1870}
1871
1872fn load_parquet_inferred(
1873 path: &Path,
1874 requested_columns: &[String],
1875 categorical_roles: &HashSet<&str>,
1876) -> Result<EncodedDataset, DataError> {
1877 use parquet::arrow::{ProjectionMask, arrow_reader::ParquetRecordBatchReaderBuilder};
1878 use rayon::prelude::*;
1879 use std::fs::File;
1880
1881 let t_open = std::time::Instant::now();
1882 let file = File::open(path).map_err(|e| DataError::ParseError {
1883 reason: format!("failed to open parquet '{}': {e}", path.display()),
1884 })?;
1885 let builder =
1886 ParquetRecordBatchReaderBuilder::try_new(file).map_err(|e| DataError::ParseError {
1887 reason: format!("failed to read parquet metadata '{}': {e}", path.display()),
1888 })?;
1889
1890 let full_schema = builder.schema().clone();
1891 let all_headers: Vec<String> = full_schema
1892 .fields()
1893 .iter()
1894 .map(|f| f.name().clone())
1895 .collect();
1896 if all_headers.is_empty() {
1897 return Err(DataError::EmptyInput {
1898 reason: "parquet file has no columns".to_string(),
1899 });
1900 }
1901 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
1902 let headers = projected_headers(&all_headers, &selected_indices);
1903 let selected_fields = selected_indices
1904 .iter()
1905 .map(|&idx| full_schema.fields()[idx].clone())
1906 .collect::<Vec<_>>();
1907 let total_rows =
1908 usize::try_from(builder.metadata().file_metadata().num_rows()).map_err(|_| {
1909 DataError::ParseError {
1910 reason: "parquet row count does not fit in memory address space".to_string(),
1911 }
1912 })?;
1913 if total_rows == 0 {
1914 return Err(DataError::EmptyInput {
1915 reason: "parquet file has no rows".to_string(),
1916 });
1917 }
1918 let projection =
1919 ProjectionMask::roots(builder.parquet_schema(), selected_indices.iter().copied());
1920 let reader =
1921 builder
1922 .with_projection(projection)
1923 .build()
1924 .map_err(|e| DataError::ParseError {
1925 reason: format!("failed to build parquet reader: {e}"),
1926 })?;
1927 let p = headers.len();
1928 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
1929 if open_ms > 100.0 {
1930 log::info!(
1931 "[DATA-LOAD] parquet_open+meta | n_headers={} | n_proj={} | {:.1}ms",
1932 all_headers.len(),
1933 p,
1934 open_ms
1935 );
1936 }
1937
1938 let t_batches = std::time::Instant::now();
1939 let is_string_col = selected_fields
1940 .iter()
1941 .map(|field| arrow_field_is_string(field.data_type()))
1942 .collect::<Vec<_>>();
1943 let forced_numeric_categorical = headers
1944 .iter()
1945 .enumerate()
1946 .map(|(j, header)| !is_string_col[j] && categorical_roles.contains(header.as_str()))
1947 .collect::<Vec<_>>();
1948 let mut values = Array2::<f64>::zeros((total_rows, p));
1949 let mut all_binary = vec![true; p];
1950 let mut categorical_encoders = (0..p)
1951 .map(|j| {
1952 (is_string_col[j] || forced_numeric_categorical[j]).then(CategoricalEncoder::default)
1953 })
1954 .collect::<Vec<_>>();
1955 let mut rows_seen = 0usize;
1956 for batch_result in reader {
1957 let batch = batch_result.map_err(|e| DataError::ParseError {
1958 reason: format!("failed to read parquet record batch: {e}"),
1959 })?;
1960 let n_rows = batch.num_rows();
1961 if rows_seen.saturating_add(n_rows) > total_rows {
1962 return Err(DataError::SchemaMismatch {
1963 reason: "parquet row count changed while reading record batches".to_string(),
1964 });
1965 }
1966
1967 let decoded_columns = values
1968 .slice_mut(s![rows_seen..rows_seen + n_rows, ..])
1969 .axis_iter_mut(Axis(1))
1970 .into_par_iter()
1971 .zip(categorical_encoders.par_iter_mut())
1972 .zip(all_binary.par_iter_mut())
1973 .enumerate()
1974 .map(|(j, ((output, encoder), column_all_binary))| {
1975 decode_arrow_batch_column_into(
1976 batch.column(j).as_ref(),
1977 rows_seen,
1978 &headers[j],
1979 is_string_col[j],
1980 output,
1981 encoder.as_mut(),
1982 column_all_binary,
1983 )
1984 })
1985 .collect::<Vec<_>>();
1986
1987 for decoded in decoded_columns {
1992 decoded?;
1993 }
1994 rows_seen += n_rows;
1995 }
1996
1997 if rows_seen != total_rows {
1998 return Err(DataError::SchemaMismatch {
1999 reason: format!(
2000 "parquet metadata reports {total_rows} rows but record batches yielded {rows_seen}"
2001 ),
2002 });
2003 }
2004 let batches_ms = t_batches.elapsed().as_secs_f64() * 1000.0;
2005 if batches_ms > 100.0 {
2006 log::info!(
2007 "[DATA-LOAD] parquet_batches_decode | n_rows={} | n_cols={} | {:.1}ms",
2008 total_rows,
2009 p,
2010 batches_ms
2011 );
2012 }
2013 let t_schema = std::time::Instant::now();
2014 let mut levels = vec![Vec::<String>::new(); p];
2020 for (j, encoder) in categorical_encoders.into_iter().enumerate() {
2021 if let Some(encoder) = encoder {
2022 let order = if forced_numeric_categorical[j] {
2023 LevelOrder::Canonical
2024 } else {
2025 LevelOrder::Encounter
2026 };
2027 levels[j] = encoder.finish(values.column_mut(j), order);
2028 }
2029 }
2030 let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
2031 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2032 for j in 0..p {
2033 let kind = if is_string_col[j] || forced_numeric_categorical[j] {
2034 ColumnKindTag::Categorical
2035 } else if all_binary[j] {
2036 ColumnKindTag::Binary
2037 } else {
2038 ColumnKindTag::Continuous
2039 };
2040 column_kinds.push(kind);
2041 schema_cols.push(SchemaColumn {
2042 name: headers[j].clone(),
2043 kind,
2044 levels: std::mem::take(&mut levels[j]),
2045 });
2046 }
2047 let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
2048 if schema_ms > 100.0 {
2049 let n_cat = column_kinds
2050 .iter()
2051 .filter(|k| matches!(k, ColumnKindTag::Categorical))
2052 .count();
2053 log::info!(
2054 "[DATA-LOAD] parquet_finalize_schema | n_cols={} | n_cat={} | {:.1}ms",
2055 p,
2056 n_cat,
2057 schema_ms
2058 );
2059 }
2060
2061 Ok(EncodedDataset {
2062 headers,
2063 values,
2064 schema: DataSchema {
2065 columns: schema_cols,
2066 },
2067 column_kinds,
2068 })
2069}
2070
2071fn load_parquet_with_schema(
2072 path: &Path,
2073 schema: &DataSchema,
2074 unseen_policy: UnseenCategoryPolicy,
2075 requested_columns: &[String],
2076) -> Result<EncodedDataset, DataError> {
2077 let inferred = load_parquet_inferred(path, requested_columns, &HashSet::new())?;
2081 let p = inferred.headers.len();
2082 let n = inferred.values.nrows();
2083
2084 let schema_byname: HashMap<&str, &SchemaColumn> = schema
2085 .columns
2086 .iter()
2087 .map(|c| (c.name.as_str(), c))
2088 .collect();
2089
2090 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2091 let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
2092 let mut values = inferred.values;
2093
2094 for j in 0..p {
2095 let name = &inferred.headers[j];
2096 if let Some(sc) = schema_byname.get(name.as_str()) {
2097 column_kinds.push(sc.kind);
2098 schema_cols.push((*sc).clone());
2099
2100 match sc.kind {
2101 ColumnKindTag::Continuous => {
2102 if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2103 return Err(DataError::SchemaMismatch {
2104 reason: format!(
2105 "column '{}' is continuous in schema but parquet column is string/categorical",
2106 name
2107 ),
2108 });
2109 }
2110 }
2111 ColumnKindTag::Binary => {
2112 if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2113 return Err(DataError::SchemaMismatch {
2114 reason: format!(
2115 "column '{}' is binary in schema but parquet column is string/categorical",
2116 name
2117 ),
2118 });
2119 }
2120 if let Some(row) = values.column(j).iter().position(|value| {
2122 value.is_finite()
2123 && (*value - 0.0).abs() >= 1e-12
2124 && (*value - 1.0).abs() >= 1e-12
2125 }) {
2126 return Err(DataError::SchemaMismatch {
2127 reason: format!(
2128 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2129 name,
2130 row + 1,
2131 values[[row, j]]
2132 ),
2133 });
2134 }
2135 }
2136 ColumnKindTag::Categorical => {
2137 if !matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2138 return Err(DataError::SchemaMismatch {
2139 reason: format!(
2140 "column '{}' is categorical in schema but parquet column is numeric",
2141 name
2142 ),
2143 });
2144 }
2145 let inferred_col = &inferred.schema.columns[j];
2146 let schema_level_map: HashMap<&str, f64> = sc
2148 .levels
2149 .iter()
2150 .enumerate()
2151 .map(|(idx, v)| (v.as_str(), idx as f64))
2152 .collect();
2153 let inferred_to_schema: Vec<f64> = inferred_col
2154 .levels
2155 .iter()
2156 .map(|lv| {
2157 schema_level_map
2158 .get(lv.as_str())
2159 .copied()
2160 .or_else(|| unseen_policy.unseen_code_for(name, sc.levels.len()))
2161 .ok_or_else(|| DataError::SchemaMismatch {
2162 reason: format!(
2163 "unseen level '{}' in categorical column '{}'",
2164 lv, name
2165 ),
2166 })
2167 })
2168 .collect::<Result<Vec<_>, _>>()?;
2169 for i in 0..n {
2170 let old_code = values[[i, j]] as usize;
2171 if old_code >= inferred_to_schema.len() {
2172 let Some(unseen_code) =
2173 unseen_policy.unseen_code_for(name, sc.levels.len())
2174 else {
2175 return Err(DataError::SchemaMismatch {
2176 reason: format!(
2177 "unseen categorical code at row {}, column '{}'",
2178 i + 1,
2179 name
2180 ),
2181 });
2182 };
2183 values[[i, j]] = unseen_code;
2184 continue;
2185 }
2186 values[[i, j]] = inferred_to_schema[old_code];
2187 }
2188 }
2189 }
2190 } else {
2191 column_kinds.push(inferred.column_kinds[j]);
2193 schema_cols.push(inferred.schema.columns[j].clone());
2194 }
2195 }
2196
2197 Ok(EncodedDataset {
2198 headers: inferred.headers,
2199 values,
2200 schema: DataSchema {
2201 columns: schema_cols,
2202 },
2203 column_kinds,
2204 })
2205}
2206
2207pub fn encode_recordswith_inferred_schema(
2208 headers: Vec<String>,
2209 records: Vec<StringRecord>,
2210) -> Result<EncodedDataset, String> {
2211 if records.is_empty() {
2212 return Err(DataError::EmptyInput {
2213 reason: "table data cannot be empty".to_string(),
2214 }
2215 .into());
2216 }
2217 let schema_cols = headers
2223 .par_iter()
2224 .enumerate()
2225 .map(|(j, name)| infer_schema_column(name, &records, j).map_err(String::from))
2226 .collect::<Result<Vec<SchemaColumn>, String>>()?;
2227 let schema = DataSchema {
2228 columns: schema_cols,
2229 };
2230 encode_recordswith_schema(headers, records, &schema, UnseenCategoryPolicy::Error)
2231}
2232
2233pub fn encode_recordswith_schema(
2234 headers: Vec<String>,
2235 records: Vec<StringRecord>,
2236 schema: &DataSchema,
2237 unseen_policy: UnseenCategoryPolicy,
2238) -> Result<EncodedDataset, String> {
2239 let n = records.len();
2240 if n == 0 {
2241 return Err(DataError::EmptyInput {
2242 reason: "table data cannot be empty".to_string(),
2243 }
2244 .into());
2245 }
2246 let p = headers.len();
2247 if p == 0 {
2248 return Err(DataError::EmptyInput {
2249 reason: "table data must have at least one header column".to_string(),
2250 }
2251 .into());
2252 }
2253 for (i, rec) in records.iter().enumerate() {
2260 if rec.len() != p {
2261 return Err(DataError::SchemaMismatch {
2262 reason: format!(
2263 "row width mismatch at row {}: got {} fields, expected {} (one per header)",
2264 i + 1,
2265 rec.len(),
2266 p
2267 ),
2268 }
2269 .into());
2270 }
2271 }
2272 let schema_byname: HashMap<&str, &SchemaColumn> = schema
2273 .columns
2274 .iter()
2275 .map(|c| (c.name.as_str(), c))
2276 .collect();
2277
2278 let encoded_columns = headers
2284 .par_iter()
2285 .enumerate()
2286 .map(|(j, name)| {
2287 let inferred_for_extra;
2288 let col_schema = if let Some(s) = schema_byname.get(name.as_str()) {
2289 *s
2290 } else {
2291 inferred_for_extra =
2292 infer_schema_column(name, &records, j).map_err(String::from)?;
2293 &inferred_for_extra
2294 };
2295 let column = encode_one_column(name, &records, j, col_schema, &unseen_policy)?;
2296 Ok::<(ColumnKindTag, Vec<f64>), String>((col_schema.kind, column))
2297 })
2298 .collect::<Result<Vec<(ColumnKindTag, Vec<f64>)>, String>>()?;
2299
2300 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2301 let mut values = Array2::<f64>::zeros((n, p));
2302 for (j, (kind, column)) in encoded_columns.into_iter().enumerate() {
2303 column_kinds.push(kind);
2304 values
2305 .column_mut(j)
2306 .assign(&ndarray::ArrayView1::from(&column));
2307 }
2308
2309 Ok(EncodedDataset {
2310 headers,
2311 values,
2312 schema: schema.clone(),
2313 column_kinds,
2314 })
2315}
2316
2317fn encode_one_column(
2324 name: &str,
2325 records: &[StringRecord],
2326 j: usize,
2327 col_schema: &SchemaColumn,
2328 unseen_policy: &UnseenCategoryPolicy,
2329) -> Result<Vec<f64>, String> {
2330 let level_map = if matches!(col_schema.kind, ColumnKindTag::Categorical) {
2331 Some(
2332 col_schema
2333 .levels
2334 .iter()
2335 .enumerate()
2336 .map(|(idx, v)| (v.as_str(), idx as f64))
2337 .collect::<HashMap<_, _>>(),
2338 )
2339 } else {
2340 None
2341 };
2342
2343 let mut column = Vec::<f64>::with_capacity(records.len());
2344 for (i, rec) in records.iter().enumerate() {
2345 let raw = rec
2346 .get(j)
2347 .ok_or_else(|| {
2348 String::from(DataError::SchemaMismatch {
2349 reason: format!("missing field at row {}, col {}", i + 1, j + 1),
2350 })
2351 })?
2352 .trim();
2353 if raw.is_empty() {
2354 return Err(DataError::EmptyInput {
2355 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2356 }
2357 .into());
2358 }
2359 let val = match col_schema.kind {
2360 ColumnKindTag::Continuous if is_missing_marker(raw) => f64::NAN,
2362 ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
2363 String::from(DataError::SchemaMismatch {
2364 reason: format!(
2365 "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
2366 name,
2367 i + 1,
2368 raw,
2369 err
2370 ),
2371 })
2372 })?,
2373 ColumnKindTag::Binary if is_missing_marker(raw) => f64::NAN,
2374 ColumnKindTag::Binary => {
2375 let v = raw.parse::<f64>().map_err(|err| {
2376 String::from(DataError::SchemaMismatch {
2377 reason: format!(
2378 "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
2379 name,
2380 i + 1,
2381 raw,
2382 err
2383 ),
2384 })
2385 })?;
2386 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2387 return Err(DataError::SchemaMismatch {
2388 reason: format!(
2389 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2390 name,
2391 i + 1,
2392 v
2393 ),
2394 }
2395 .into());
2396 }
2397 v
2398 }
2399 ColumnKindTag::Categorical => {
2400 let map = level_map.as_ref().ok_or_else(|| {
2401 String::from(DataError::EncodingFailure {
2402 reason: "internal categorical schema map missing".to_string(),
2403 })
2404 })?;
2405 match map.get(raw) {
2406 Some(v) => *v,
2407 None => unseen_policy
2408 .unseen_code_for(name, col_schema.levels.len())
2409 .ok_or_else(|| {
2410 String::from(DataError::SchemaMismatch {
2411 reason: format!(
2412 "unseen level '{}' in categorical column '{}' at row {}; allowed levels: {}",
2413 raw,
2414 name,
2415 i + 1,
2416 col_schema.levels.join(",")
2417 ),
2418 })
2419 })?,
2420 }
2421 }
2422 };
2423 if !val.is_finite() && !is_missing_marker(raw) {
2425 return Err(DataError::InvalidValue {
2426 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2427 }
2428 .into());
2429 }
2430 column.push(val);
2431 }
2432 Ok(column)
2433}
2434
2435fn infer_schema_column(
2436 name: &str,
2437 records: &[StringRecord],
2438 col_idx: usize,
2439) -> Result<SchemaColumn, DataError> {
2440 let mut all_numeric = true;
2441 let mut all_binary = true;
2442 let mut saw_numeric = false;
2443 let mut levels = Vec::<String>::new();
2444 let mut level_index = HashMap::<String, usize>::new();
2445 let mut missing_markers = Vec::<String>::new();
2449 for (i, rec) in records.iter().enumerate() {
2450 let raw = rec
2451 .get(col_idx)
2452 .ok_or_else(|| DataError::SchemaMismatch {
2453 reason: format!("missing field at row {}, col {}", i + 1, col_idx + 1),
2454 })?
2455 .trim();
2456 if raw.is_empty() {
2457 return Err(DataError::EmptyInput {
2458 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2459 });
2460 }
2461 if is_missing_marker(raw) {
2462 missing_markers.push(raw.to_string());
2463 continue;
2464 }
2465 if let Ok(v) = raw.parse::<f64>() {
2466 saw_numeric = true;
2467 if !v.is_finite() {
2468 return Err(DataError::InvalidValue {
2469 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2470 });
2471 }
2472 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2473 all_binary = false;
2474 }
2475 } else {
2476 all_numeric = false;
2477 all_binary = false;
2478 level_index.entry(raw.to_string()).or_insert_with(|| {
2479 let idx = levels.len();
2480 levels.push(raw.to_string());
2481 idx
2482 });
2483 }
2484 }
2485 let numeric_column = all_numeric && saw_numeric;
2498 if !numeric_column {
2499 for marker in missing_markers {
2500 level_index.entry(marker.clone()).or_insert_with(|| {
2501 let idx = levels.len();
2502 levels.push(marker);
2503 idx
2504 });
2505 }
2506 }
2507 let kind = if numeric_column {
2508 if all_binary {
2509 ColumnKindTag::Binary
2510 } else {
2511 ColumnKindTag::Continuous
2512 }
2513 } else {
2514 ColumnKindTag::Categorical
2515 };
2516 if matches!(kind, ColumnKindTag::Categorical) {
2521 sort_levels_canonical(&mut levels);
2522 }
2523 Ok(SchemaColumn {
2524 name: name.to_string(),
2525 kind,
2526 levels: if matches!(kind, ColumnKindTag::Categorical) {
2527 levels
2528 } else {
2529 Vec::new()
2530 },
2531 })
2532}
2533
2534pub fn infer_and_encode_column_major(
2547 name: &str,
2548 column: &[&str],
2549 col_index: usize,
2550) -> Result<(SchemaColumn, Vec<f64>), String> {
2551 if column.is_empty() {
2552 return Err(DataError::EmptyInput {
2553 reason: "table data cannot be empty".to_string(),
2554 }
2555 .into());
2556 }
2557 let force_categorical = column.iter().any(|c| strip_categorical_sentinel(c).1);
2562 let mut all_numeric = !force_categorical;
2563 let mut all_binary = !force_categorical;
2564 let mut levels = Vec::<String>::new();
2565 let mut level_index = HashMap::<String, usize>::new();
2566 let mut trimmed = Vec::<&str>::with_capacity(column.len());
2567 let mut parsed = Vec::<Option<f64>>::with_capacity(column.len());
2574 let mut saw_numeric = false;
2575 let mut missing_positions = Vec::<usize>::new();
2576 for (i, raw_field) in column.iter().enumerate() {
2577 let (raw, _) = strip_categorical_sentinel(raw_field);
2580 let raw = raw.trim();
2581 if raw.is_empty() {
2582 return Err(DataError::EmptyInput {
2583 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2584 }
2585 .into());
2586 }
2587 if !force_categorical {
2590 if is_missing_marker(raw) {
2594 missing_positions.push(i);
2595 parsed.push(Some(f64::NAN));
2596 trimmed.push(raw);
2597 continue;
2598 }
2599 if let Ok(v) = raw.parse::<f64>() {
2600 saw_numeric = true;
2601 if !v.is_finite() {
2602 return Err(DataError::InvalidValue {
2603 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2604 }
2605 .into());
2606 }
2607 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2608 all_binary = false;
2609 }
2610 parsed.push(Some(v));
2611 trimmed.push(raw);
2612 continue;
2613 }
2614 all_numeric = false;
2615 all_binary = false;
2616 }
2617 level_index.entry(raw.to_string()).or_insert_with(|| {
2618 let idx = levels.len();
2619 levels.push(raw.to_string());
2620 idx
2621 });
2622 parsed.push(None);
2623 trimmed.push(raw);
2624 }
2625 let numeric_column = all_numeric && saw_numeric;
2630 if !numeric_column {
2631 for &i in &missing_positions {
2632 let raw = trimmed[i];
2633 level_index.entry(raw.to_string()).or_insert_with(|| {
2634 let idx = levels.len();
2635 levels.push(raw.to_string());
2636 idx
2637 });
2638 parsed[i] = None;
2639 }
2640 }
2641 let kind = if numeric_column {
2642 if all_binary {
2643 ColumnKindTag::Binary
2644 } else {
2645 ColumnKindTag::Continuous
2646 }
2647 } else {
2648 ColumnKindTag::Categorical
2649 };
2650 if matches!(kind, ColumnKindTag::Categorical) {
2664 sort_levels_canonical(&mut levels);
2665 }
2666 let schema = SchemaColumn {
2667 name: name.to_string(),
2668 kind,
2669 levels: if matches!(kind, ColumnKindTag::Categorical) {
2670 levels
2671 } else {
2672 Vec::new()
2673 },
2674 };
2675
2676 let level_map = if matches!(kind, ColumnKindTag::Categorical) {
2677 Some(
2678 schema
2679 .levels
2680 .iter()
2681 .enumerate()
2682 .map(|(idx, v)| (v.as_str(), idx as f64))
2683 .collect::<HashMap<_, _>>(),
2684 )
2685 } else {
2686 None
2687 };
2688
2689 let mut values = Vec::<f64>::with_capacity(trimmed.len());
2690 for (i, raw) in trimmed.iter().enumerate() {
2691 let raw = *raw;
2692 let val = match kind {
2693 ColumnKindTag::Continuous => parsed[i].ok_or_else(|| {
2697 String::from(DataError::EncodingFailure {
2698 reason: format!(
2699 "internal: continuous column '{}' lost its parsed value at row {} (col {})",
2700 name,
2701 i + 1,
2702 col_index
2703 ),
2704 })
2705 })?,
2706 ColumnKindTag::Binary => {
2707 let v = parsed[i].ok_or_else(|| {
2708 String::from(DataError::EncodingFailure {
2709 reason: format!(
2710 "internal: binary column '{}' lost its parsed value at row {} (col {})",
2711 name,
2712 i + 1,
2713 col_index
2714 ),
2715 })
2716 })?;
2717 if v.is_finite() && (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2720 return Err(DataError::SchemaMismatch {
2721 reason: format!(
2722 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2723 name,
2724 i + 1,
2725 v
2726 ),
2727 }
2728 .into());
2729 }
2730 v
2731 }
2732 ColumnKindTag::Categorical => {
2733 let map = level_map.as_ref().ok_or_else(|| {
2734 String::from(DataError::EncodingFailure {
2735 reason: "internal categorical schema map missing".to_string(),
2736 })
2737 })?;
2738 *map.get(raw).ok_or_else(|| {
2739 String::from(DataError::EncodingFailure {
2740 reason: format!(
2741 "internal: level '{}' missing from freshly built map for column '{}' (col {})",
2742 raw, name, col_index
2743 ),
2744 })
2745 })?
2746 }
2747 };
2748 if !val.is_finite() && !is_missing_marker(raw) {
2750 return Err(DataError::InvalidValue {
2751 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2752 }
2753 .into());
2754 }
2755 values.push(val);
2756 }
2757 Ok((schema, values))
2758}
2759
2760#[cfg(test)]
2761mod missing_value_inference_tests {
2762 use super::*;
2763
2764 fn rows(cells: &[&[&str]]) -> Vec<StringRecord> {
2765 cells
2766 .iter()
2767 .map(|r| StringRecord::from(r.to_vec()))
2768 .collect()
2769 }
2770
2771 #[test]
2778 fn a_numeric_column_containing_na_can_never_infer_categorical() {
2779 let ds = encode_recordswith_inferred_schema(
2780 vec!["parker".to_string()],
2781 rows(&[&["94.4"], &["NA"], &["65.0"], &["88.0"], &["NA"]]),
2782 )
2783 .expect("encode a numeric column carrying NA");
2784
2785 assert_eq!(
2786 ds.schema.columns[0].kind,
2787 ColumnKindTag::Continuous,
2788 "a column whose present cells are all numeric is numeric-with-missing, \
2789 never a factor over its own measurements"
2790 );
2791 assert!(
2792 ds.schema.columns[0].levels.is_empty(),
2793 "measurements must not be recorded as factor levels"
2794 );
2795
2796 let col: Vec<f64> = ds.values.column(0).to_vec();
2799 assert_eq!(col[0], 94.4);
2800 assert_eq!(col[2], 65.0);
2801 assert_eq!(col[3], 88.0);
2802 assert!(col[1].is_nan() && col[4].is_nan(), "NA must encode as NaN");
2803 assert_eq!(
2804 col.iter().filter(|v| v.is_finite()).count(),
2805 3,
2806 "is_finite() must count exactly the present cells"
2807 );
2808 }
2809
2810 #[test]
2815 fn na_stays_a_level_in_a_genuinely_categorical_column() {
2816 let ds = encode_recordswith_inferred_schema(
2817 vec!["country".to_string()],
2818 rows(&[&["NA"], &["ZA"], &["BW"], &["NA"]]),
2819 )
2820 .expect("encode a categorical column whose labels include NA");
2821
2822 assert_eq!(ds.schema.columns[0].kind, ColumnKindTag::Categorical);
2823 assert!(
2824 ds.schema.columns[0].levels.iter().any(|l| l == "NA"),
2825 "NA is a country here, not missingness: levels were {:?}",
2826 ds.schema.columns[0].levels
2827 );
2828 assert!(
2829 ds.values.column(0).iter().all(|v| v.is_finite()),
2830 "a categorical column carries level codes, never NaN"
2831 );
2832 }
2833
2834 #[test]
2837 fn a_binary_column_containing_na_stays_binary_with_nan_holes() {
2838 let ds = encode_recordswith_inferred_schema(
2839 vec!["event".to_string()],
2840 rows(&[&["1"], &["NA"], &["0"], &["1"]]),
2841 )
2842 .expect("encode a binary column carrying NA");
2843
2844 assert_eq!(ds.schema.columns[0].kind, ColumnKindTag::Binary);
2845 let col: Vec<f64> = ds.values.column(0).to_vec();
2846 assert_eq!((col[0], col[2], col[3]), (1.0, 0.0, 1.0));
2847 assert!(col[1].is_nan());
2848 }
2849
2850 #[test]
2855 fn a_parsed_non_finite_literal_still_fails_loudly() {
2856 let err = encode_recordswith_inferred_schema(
2857 vec!["x".to_string()],
2858 rows(&[&["1.0"], &["inf"], &["2.0"]]),
2859 )
2860 .expect_err("a literal infinity is a data error, not a missing value");
2861 assert!(
2862 err.contains("non-finite"),
2863 "expected the non-finite guard, got: {err}"
2864 );
2865 }
2866}
2867
2868#[cfg(test)]
2869mod tests {
2870 use super::*;
2871 use arrow::array::ArrayRef;
2872 use arrow::datatypes::{Field, Schema};
2873 use arrow::error::ArrowError;
2874 use arrow::record_batch::{RecordBatch, RecordBatchIterator};
2875 use std::sync::Arc;
2876
2877 fn encode_single_arrow_array(array: ArrayRef) -> Result<EncodedDataset, DataError> {
2878 let schema = Arc::new(Schema::new(vec![Field::new(
2879 "source",
2880 array.data_type().clone(),
2881 true,
2882 )]));
2883 let batch = RecordBatch::try_new(schema.clone(), vec![array]).expect("record batch");
2884 let batches: Vec<Result<RecordBatch, ArrowError>> = vec![Ok(batch)];
2885 let mut reader = RecordBatchIterator::new(batches, schema);
2886 encode_arrow_record_batch_reader_with_inferred_schema(
2887 &mut reader,
2888 vec!["normalized".to_string()],
2889 )
2890 }
2891
2892 #[test]
2893 fn arrow_reader_streams_typed_columns_in_supplied_order() {
2894 use arrow::array::{
2895 Array, BooleanArray, DictionaryArray, Float32Array, Int8Array, Int32Array, Int64Array,
2896 LargeStringArray, StringArray,
2897 };
2898 use arrow::datatypes::Int8Type;
2899
2900 let string_dictionary_1: DictionaryArray<Int8Type> =
2901 vec!["beta", "alpha"].into_iter().collect();
2902 let string_dictionary_2: DictionaryArray<Int8Type> = vec!["beta"].into_iter().collect();
2903 let numeric_dictionary_1 = DictionaryArray::<Int8Type>::new(
2904 Int8Array::from(vec![0, 1]),
2905 Arc::new(Int64Array::from(vec![5, 7])),
2906 );
2907 let numeric_dictionary_2 = DictionaryArray::<Int8Type>::new(
2908 Int8Array::from(vec![0]),
2909 Arc::new(Int64Array::from(vec![5])),
2910 );
2911
2912 let schema = Arc::new(Schema::new(vec![
2913 Field::new("source_float", arrow::datatypes::DataType::Float32, false),
2914 Field::new("source_integer", arrow::datatypes::DataType::Int32, false),
2915 Field::new("source_flag", arrow::datatypes::DataType::Boolean, false),
2916 Field::new("source_utf8", arrow::datatypes::DataType::Utf8, false),
2917 Field::new(
2918 "source_large_utf8",
2919 arrow::datatypes::DataType::LargeUtf8,
2920 false,
2921 ),
2922 Field::new(
2923 "source_dictionary_string",
2924 string_dictionary_1.data_type().clone(),
2925 false,
2926 ),
2927 Field::new(
2928 "source_dictionary_number",
2929 numeric_dictionary_1.data_type().clone(),
2930 false,
2931 ),
2932 ]));
2933 let batch_1 = RecordBatch::try_new(
2934 schema.clone(),
2935 vec![
2936 Arc::new(Float32Array::from(vec![1.5, 2.5])) as ArrayRef,
2937 Arc::new(Int32Array::from(vec![0, 1])),
2938 Arc::new(BooleanArray::from(vec![true, false])),
2939 Arc::new(StringArray::from(vec!["item10", "item2"])),
2940 Arc::new(LargeStringArray::from(vec!["z", "a"])),
2941 Arc::new(string_dictionary_1),
2942 Arc::new(numeric_dictionary_1),
2943 ],
2944 )
2945 .expect("first record batch");
2946 let batch_2 = RecordBatch::try_new(
2947 schema.clone(),
2948 vec![
2949 Arc::new(Float32Array::from(vec![-4.0])) as ArrayRef,
2950 Arc::new(Int32Array::from(vec![1])),
2951 Arc::new(BooleanArray::from(vec![true])),
2952 Arc::new(StringArray::from(vec!["item1"])),
2953 Arc::new(LargeStringArray::from(vec!["z"])),
2954 Arc::new(string_dictionary_2),
2955 Arc::new(numeric_dictionary_2),
2956 ],
2957 )
2958 .expect("second record batch");
2959 let batches: Vec<Result<RecordBatch, ArrowError>> = vec![Ok(batch_1), Ok(batch_2)];
2960 let mut reader = RecordBatchIterator::new(batches, schema);
2961 let headers = [
2962 "float",
2963 "integer",
2964 "flag",
2965 "utf8",
2966 "large_utf8",
2967 "dictionary_string",
2968 "dictionary_number",
2969 ]
2970 .map(str::to_string)
2971 .to_vec();
2972
2973 let dataset =
2974 encode_arrow_record_batch_reader_with_inferred_schema(&mut reader, headers.clone())
2975 .expect("Arrow stream should encode");
2976
2977 assert_eq!(dataset.headers, headers);
2978 assert_eq!(
2979 dataset.column_kinds,
2980 vec![
2981 ColumnKindTag::Continuous,
2982 ColumnKindTag::Binary,
2983 ColumnKindTag::Binary,
2984 ColumnKindTag::Categorical,
2985 ColumnKindTag::Categorical,
2986 ColumnKindTag::Categorical,
2987 ColumnKindTag::Continuous,
2988 ]
2989 );
2990 assert_eq!(
2991 dataset.values,
2992 ndarray::arr2(&[
2993 [1.5, 0.0, 1.0, 2.0, 1.0, 1.0, 5.0],
2994 [2.5, 1.0, 0.0, 1.0, 0.0, 0.0, 7.0],
2995 [-4.0, 1.0, 1.0, 0.0, 1.0, 1.0, 5.0],
2996 ])
2997 );
2998 assert_eq!(
2999 dataset.schema.columns[3].levels,
3000 vec!["item1", "item2", "item10"]
3001 );
3002 assert_eq!(dataset.schema.columns[4].levels, vec!["a", "z"]);
3003 assert_eq!(dataset.schema.columns[5].levels, vec!["alpha", "beta"]);
3004 assert!(
3005 dataset
3006 .schema
3007 .columns
3008 .iter()
3009 .zip(dataset.headers.iter())
3010 .all(|(column, header)| column.name == *header)
3011 );
3012 }
3013
3014 #[test]
3015 fn arrow_reader_rejects_empty_duplicate_and_mismatched_headers() {
3016 let schema = Arc::new(Schema::new(vec![Field::new(
3017 "source",
3018 arrow::datatypes::DataType::Int32,
3019 false,
3020 )]));
3021
3022 let mut empty_name_reader = RecordBatchIterator::new(
3023 Vec::<Result<RecordBatch, ArrowError>>::new(),
3024 schema.clone(),
3025 );
3026 let empty_name = encode_arrow_record_batch_reader_with_inferred_schema(
3027 &mut empty_name_reader,
3028 vec![" ".to_string()],
3029 )
3030 .expect_err("blank header should fail");
3031 assert!(matches!(empty_name, DataError::EmptyInput { .. }));
3032
3033 let mut duplicate_reader = RecordBatchIterator::new(
3034 Vec::<Result<RecordBatch, ArrowError>>::new(),
3035 Arc::new(Schema::new(vec![
3036 Field::new("a", arrow::datatypes::DataType::Int32, false),
3037 Field::new("b", arrow::datatypes::DataType::Int32, false),
3038 ])),
3039 );
3040 let duplicate = encode_arrow_record_batch_reader_with_inferred_schema(
3041 &mut duplicate_reader,
3042 vec!["x".to_string(), "x".to_string()],
3043 )
3044 .expect_err("duplicate header should fail");
3045 assert!(matches!(duplicate, DataError::SchemaMismatch { .. }));
3046
3047 let mut mismatch_reader =
3048 RecordBatchIterator::new(Vec::<Result<RecordBatch, ArrowError>>::new(), schema);
3049 let mismatch = encode_arrow_record_batch_reader_with_inferred_schema(
3050 &mut mismatch_reader,
3051 vec!["x".to_string(), "y".to_string()],
3052 )
3053 .expect_err("header count mismatch should fail");
3054 assert!(matches!(mismatch, DataError::SchemaMismatch { .. }));
3055 }
3056
3057 #[test]
3058 fn arrow_reader_reports_typed_null_nonfinite_and_unsupported_errors() {
3059 use arrow::array::{Date32Array, DictionaryArray, Float64Array, Int8Array, StringArray};
3060 use arrow::datatypes::Int8Type;
3061
3062 let null_numeric =
3063 encode_single_arrow_array(Arc::new(Float64Array::from(vec![Some(1.0), None])))
3064 .expect_err("numeric null should fail");
3065 assert!(matches!(&null_numeric, DataError::InvalidValue { .. }));
3066 assert!(null_numeric.to_string().contains("null value at row 2"));
3067
3068 let null_dictionary = DictionaryArray::<Int8Type>::new(
3069 Int8Array::from(vec![0, 1]),
3070 Arc::new(StringArray::from(vec![Some("present"), None])),
3071 );
3072 let logical_null = encode_single_arrow_array(Arc::new(null_dictionary))
3073 .expect_err("null dictionary value should fail");
3074 assert!(matches!(&logical_null, DataError::InvalidValue { .. }));
3075 assert!(logical_null.to_string().contains("null value at row 2"));
3076
3077 let nonfinite = encode_single_arrow_array(Arc::new(Float64Array::from(vec![f64::NAN])))
3078 .expect_err("NaN should fail");
3079 assert!(matches!(&nonfinite, DataError::InvalidValue { .. }));
3080 assert!(nonfinite.to_string().contains("non-finite value"));
3081
3082 let unsupported = encode_single_arrow_array(Arc::new(Date32Array::from(vec![1])))
3083 .expect_err("date column should fail");
3084 assert!(matches!(&unsupported, DataError::InvalidValue { .. }));
3085 assert!(
3086 unsupported
3087 .to_string()
3088 .contains("unsupported Arrow column type")
3089 );
3090 }
3091
3092 #[test]
3093 fn encode_records_rejects_empty_input() {
3094 let headers = vec!["x".to_string()];
3095 let schema = DataSchema {
3096 columns: vec![SchemaColumn {
3097 name: "x".to_string(),
3098 kind: ColumnKindTag::Continuous,
3099 levels: Vec::new(),
3100 }],
3101 };
3102
3103 let err = encode_recordswith_inferred_schema(headers.clone(), Vec::new())
3104 .expect_err("empty inferred records should error");
3105 assert_eq!(err, "table data cannot be empty");
3106
3107 let err =
3108 encode_recordswith_schema(headers, Vec::new(), &schema, UnseenCategoryPolicy::Error)
3109 .expect_err("empty schema-guided records should error");
3110 assert_eq!(err, "table data cannot be empty");
3111 }
3112
3113 #[test]
3114 fn column_major_matches_record_driven_inferred_encode() {
3115 let headers = vec!["cont".to_string(), "bin".to_string(), "cat".to_string()];
3120 let raw_rows = vec![
3121 vec!["1.5", "0", "a"],
3122 vec!["2.0", "1", "b"],
3123 vec!["-3.25", "1", "a"],
3124 vec!["0.0", "0", "c"],
3125 ];
3126 let records: Vec<StringRecord> = raw_rows
3127 .iter()
3128 .map(|r| StringRecord::from(r.clone()))
3129 .collect();
3130 let record_ds = encode_recordswith_inferred_schema(headers.clone(), records)
3131 .expect("record-driven encode");
3132
3133 for (j, name) in headers.iter().enumerate() {
3134 let column: Vec<&str> = raw_rows.iter().map(|r| r[j]).collect();
3135 let (schema_col, values) =
3136 infer_and_encode_column_major(name, &column, j + 1).expect("column-major encode");
3137 assert_eq!(schema_col.kind, record_ds.schema.columns[j].kind);
3138 assert_eq!(schema_col.levels, record_ds.schema.columns[j].levels);
3139 for (i, v) in values.iter().enumerate() {
3140 assert_eq!(*v, record_ds.values[[i, j]], "row {i} col {name}");
3141 }
3142 }
3143 }
3144
3145 #[test]
3146 fn encode_records_can_encode_unseen_named_categorical_column() {
3147 let schema = DataSchema {
3148 columns: vec![
3149 SchemaColumn {
3150 name: "g".to_string(),
3151 kind: ColumnKindTag::Categorical,
3152 levels: vec!["a".to_string(), "b".to_string()],
3153 },
3154 SchemaColumn {
3155 name: "x".to_string(),
3156 kind: ColumnKindTag::Categorical,
3157 levels: vec!["low".to_string(), "high".to_string()],
3158 },
3159 ],
3160 };
3161 let headers = vec!["g".to_string(), "x".to_string()];
3162 let records = vec![StringRecord::from(vec!["new-group", "low"])];
3163 let policy =
3164 UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
3165
3166 let ds =
3167 encode_recordswith_schema(headers, records, &schema, policy).expect("encoded dataset");
3168
3169 assert_eq!(ds.values[[0, 0]], 2.0);
3170 assert_eq!(ds.values[[0, 1]], 0.0);
3171 }
3172
3173 #[test]
3174 fn categorical_encoder_consumes_labels_and_remaps_canonically() {
3175 use ndarray::Array1;
3176
3177 let mut encoder = CategoricalEncoder::default();
3178 let mut encoded = Array1::from_vec(
3179 ["item10", "item2", "item1", "item2"]
3180 .into_iter()
3181 .map(|label| encoder.encode(label) as f64)
3182 .collect(),
3183 );
3184
3185 let levels = encoder.finish(encoded.view_mut(), LevelOrder::Canonical);
3186
3187 assert_eq!(levels, vec!["item1", "item2", "item10"]);
3188 assert_eq!(encoded.to_vec(), vec![2.0, 1.0, 0.0, 1.0]);
3189 }
3190
3191 #[test]
3192 fn complete_delimited_schema_encodes_projected_rows_directly() {
3193 let dir = tempfile::tempdir().expect("tempdir");
3194 let path = dir.path().join("schema_direct.csv");
3195 std::fs::write(
3196 &path,
3197 "y,group,flag,unused\n1.5,b,0,first\n2.5,a,1,second\n",
3198 )
3199 .expect("write csv");
3200 let schema = DataSchema {
3201 columns: vec![
3202 SchemaColumn {
3203 name: "group".to_string(),
3204 kind: ColumnKindTag::Categorical,
3205 levels: vec!["a".to_string(), "b".to_string()],
3206 },
3207 SchemaColumn {
3208 name: "flag".to_string(),
3209 kind: ColumnKindTag::Binary,
3210 levels: Vec::new(),
3211 },
3212 SchemaColumn {
3213 name: "y".to_string(),
3214 kind: ColumnKindTag::Continuous,
3215 levels: Vec::new(),
3216 },
3217 ],
3218 };
3219
3220 let loaded = load_datasetwith_schema_projected(
3221 &path,
3222 &schema,
3223 UnseenCategoryPolicy::Error,
3224 &["y".to_string(), "group".to_string(), "flag".to_string()],
3225 )
3226 .expect("schema-guided projected load");
3227
3228 assert_eq!(loaded.headers, vec!["y", "group", "flag"]);
3229 assert_eq!(loaded.values.row(0).to_vec(), vec![1.5, 1.0, 0.0]);
3230 assert_eq!(loaded.values.row(1).to_vec(), vec![2.5, 0.0, 1.0]);
3231 assert_eq!(
3232 loaded.column_kinds,
3233 vec![
3234 ColumnKindTag::Continuous,
3235 ColumnKindTag::Categorical,
3236 ColumnKindTag::Binary,
3237 ]
3238 );
3239 }
3240
3241 #[test]
3242 fn direct_parquet_decoder_preserves_string_encounter_order() {
3243 use arrow::array::DictionaryArray;
3244 use arrow::datatypes::Int8Type;
3245 use ndarray::Array1;
3246
3247 let dictionary: DictionaryArray<Int8Type> =
3248 vec!["beta", "alpha", "beta"].into_iter().collect();
3249 let mut encoded = Array1::<f64>::zeros(dictionary.len());
3250 let mut encoder = CategoricalEncoder::default();
3251 let mut all_binary = true;
3252
3253 decode_arrow_batch_column_into(
3254 &dictionary,
3255 0,
3256 "group",
3257 true,
3258 encoded.view_mut(),
3259 Some(&mut encoder),
3260 &mut all_binary,
3261 )
3262 .expect("dictionary strings decode directly");
3263 let levels = encoder.finish(encoded.view_mut(), LevelOrder::Encounter);
3264
3265 assert_eq!(levels, vec!["beta", "alpha"]);
3266 assert_eq!(encoded.to_vec(), vec![0.0, 1.0, 0.0]);
3267 }
3268
3269 #[test]
3270 fn numeric_valued_dictionary_column_classifies_and_decodes_as_numeric() {
3271 use arrow::array::{Array, ArrayRef, DictionaryArray, Int8Array, Int64Array};
3281 use arrow::datatypes::{DataType, Int8Type};
3282 use std::sync::Arc;
3283
3284 let keys = Int8Array::from(vec![0i8, 1, 0, 1, 0]);
3286 let dict_values: ArrayRef = Arc::new(Int64Array::from(vec![5i64, 7]));
3287 let dict: DictionaryArray<Int8Type> = DictionaryArray::new(keys, dict_values);
3288
3289 assert!(matches!(dict.data_type(), DataType::Dictionary(_, _)));
3292 assert!(
3293 !arrow_field_is_string(dict.data_type()),
3294 "Dictionary(Int8, Int64) must not be treated as a string column"
3295 );
3296
3297 let str_dict: DictionaryArray<Int8Type> = vec!["a", "b", "a"].into_iter().collect();
3299 assert!(
3300 arrow_field_is_string(str_dict.data_type()),
3301 "Dictionary(Int8, Utf8) must remain a string column"
3302 );
3303
3304 let mut decoded = ndarray::Array1::<f64>::zeros(dict.len());
3308 let mut all_binary = true;
3309 decode_arrow_batch_column_into(
3310 &dict,
3311 0,
3312 "x",
3313 false,
3314 decoded.view_mut(),
3315 None,
3316 &mut all_binary,
3317 )
3318 .expect("numeric dictionary column should decode as numeric");
3319 assert_eq!(decoded.to_vec(), vec![5.0, 7.0, 5.0, 7.0, 5.0]);
3320 assert!(!all_binary);
3321
3322 use arrow::datatypes::{Field, Schema};
3327 use arrow::record_batch::RecordBatch;
3328 use parquet::arrow::ArrowWriter;
3329
3330 let arrow_schema = Arc::new(Schema::new(vec![Field::new(
3331 "x",
3332 dict.data_type().clone(),
3333 false,
3334 )]));
3335 let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(dict.clone())])
3336 .expect("record batch with a dictionary numeric column");
3337
3338 let dir = tempfile::tempdir().expect("tempdir");
3339 let path = dir.path().join("dict_numeric.parquet");
3340 {
3341 let file = std::fs::File::create(&path).expect("create parquet");
3342 let mut writer =
3343 ArrowWriter::try_new(file, arrow_schema, None).expect("arrow parquet writer");
3344 writer.write(&batch).expect("write batch");
3345 writer.close().expect("close writer");
3346 }
3347
3348 let inferred =
3351 load_parquet_inferred(&path, &[], &HashSet::new()).expect("inferred parquet load");
3352 assert_eq!(inferred.column_kinds, vec![ColumnKindTag::Continuous]);
3353 assert_eq!(
3354 inferred.values.column(0).to_vec(),
3355 vec![5.0, 7.0, 5.0, 7.0, 5.0]
3356 );
3357
3358 let schema = DataSchema {
3362 columns: vec![SchemaColumn {
3363 name: "x".to_string(),
3364 kind: ColumnKindTag::Continuous,
3365 levels: Vec::new(),
3366 }],
3367 };
3368 let schema_loaded =
3369 load_parquet_with_schema(&path, &schema, UnseenCategoryPolicy::Error, &[])
3370 .expect("dictionary-encoded numeric parquet must load against a Continuous schema");
3371 assert_eq!(schema_loaded.column_kinds, vec![ColumnKindTag::Continuous]);
3372 assert_eq!(
3373 schema_loaded.values.column(0).to_vec(),
3374 vec![5.0, 7.0, 5.0, 7.0, 5.0]
3375 );
3376 }
3377
3378 #[test]
3379 fn encode_records_keeps_unlisted_categorical_columns_strict() {
3380 let schema = DataSchema {
3381 columns: vec![
3382 SchemaColumn {
3383 name: "g".to_string(),
3384 kind: ColumnKindTag::Categorical,
3385 levels: vec!["a".to_string(), "b".to_string()],
3386 },
3387 SchemaColumn {
3388 name: "x".to_string(),
3389 kind: ColumnKindTag::Categorical,
3390 levels: vec!["low".to_string(), "high".to_string()],
3391 },
3392 ],
3393 };
3394 let headers = vec!["g".to_string(), "x".to_string()];
3395 let records = vec![StringRecord::from(vec!["a", "new-level"])];
3396 let policy =
3397 UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
3398
3399 let err = encode_recordswith_schema(headers, records, &schema, policy)
3400 .expect_err("ordinary categorical column should stay strict");
3401
3402 assert!(err.contains("unseen level 'new-level' in categorical column 'x'"));
3403 }
3404
3405 #[test]
3410 fn sentinel_strip_present_returns_rest_and_true() {
3411 let marked = format!("{}{}", CATEGORICAL_CELL_SENTINEL, "hello");
3412 let (rest, found) = strip_categorical_sentinel(&marked);
3413 assert_eq!(rest, "hello");
3414 assert!(found);
3415 }
3416
3417 #[test]
3418 fn sentinel_strip_absent_returns_original_and_false() {
3419 let (rest, found) = strip_categorical_sentinel("hello");
3420 assert_eq!(rest, "hello");
3421 assert!(!found);
3422 }
3423
3424 #[test]
3425 fn sentinel_strip_empty_string_returns_empty_and_false() {
3426 let (rest, found) = strip_categorical_sentinel("");
3427 assert_eq!(rest, "");
3428 assert!(!found);
3429 }
3430
3431 #[test]
3432 fn sentinel_strip_only_sentinel_returns_empty_and_true() {
3433 let marked = CATEGORICAL_CELL_SENTINEL.to_string();
3434 let (rest, found) = strip_categorical_sentinel(&marked);
3435 assert_eq!(rest, "");
3436 assert!(found);
3437 }
3438
3439 #[test]
3444 fn feature_ranges_two_columns() {
3445 let values = ndarray::arr2(&[[1.0_f64, 10.0], [3.0, 20.0], [2.0, 15.0]]);
3446 let ds = EncodedDataset {
3447 headers: vec!["a".to_string(), "b".to_string()],
3448 values,
3449 schema: DataSchema { columns: vec![] },
3450 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3451 };
3452 let ranges = ds.feature_ranges();
3453 assert_eq!(ranges.len(), 2);
3454 assert_eq!(ranges[0], (1.0, 3.0));
3455 assert_eq!(ranges[1], (10.0, 20.0));
3456 }
3457
3458 #[test]
3459 fn feature_ranges_single_row_min_equals_max() {
3460 let values = ndarray::arr2(&[[5.0_f64, -3.0]]);
3461 let ds = EncodedDataset {
3462 headers: vec!["x".to_string(), "y".to_string()],
3463 values,
3464 schema: DataSchema { columns: vec![] },
3465 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3466 };
3467 let ranges = ds.feature_ranges();
3468 assert_eq!(ranges[0], (5.0, 5.0));
3469 assert_eq!(ranges[1], (-3.0, -3.0));
3470 }
3471
3472 #[test]
3473 fn feature_ranges_all_nan_defaults_to_zero() {
3474 let values = ndarray::arr2(&[[f64::NAN], [f64::NAN]]);
3475 let ds = EncodedDataset {
3476 headers: vec!["x".to_string()],
3477 values,
3478 schema: DataSchema { columns: vec![] },
3479 column_kinds: vec![ColumnKindTag::Continuous],
3480 };
3481 let ranges = ds.feature_ranges();
3482 assert_eq!(ranges[0], (0.0, 0.0));
3483 }
3484
3485 #[test]
3490 fn column_map_indexes_by_name() {
3491 let values = ndarray::arr2(&[[0.0_f64, 1.0], [2.0, 3.0]]);
3492 let ds = EncodedDataset {
3493 headers: vec!["alpha".to_string(), "beta".to_string()],
3494 values,
3495 schema: DataSchema { columns: vec![] },
3496 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3497 };
3498 let map = ds.column_map();
3499 assert_eq!(map["alpha"], 0);
3500 assert_eq!(map["beta"], 1);
3501 assert_eq!(map.len(), 2);
3502 }
3503
3504 #[test]
3507 fn shared_prefix_identical_strings() {
3508 assert_eq!(shared_prefix("hello", "hello"), 5);
3509 }
3510
3511 #[test]
3512 fn shared_prefix_no_common_prefix() {
3513 assert_eq!(shared_prefix("abc", "xyz"), 0);
3514 }
3515
3516 #[test]
3517 fn shared_prefix_partial_match() {
3518 assert_eq!(shared_prefix("foobar", "foobaz"), 5);
3519 }
3520
3521 #[test]
3522 fn shared_prefix_one_empty() {
3523 assert_eq!(shared_prefix("", "hello"), 0);
3524 assert_eq!(shared_prefix("hello", ""), 0);
3525 }
3526
3527 #[test]
3528 fn shared_prefix_both_empty() {
3529 assert_eq!(shared_prefix("", ""), 0);
3530 }
3531
3532 #[test]
3533 fn shared_prefix_shorter_string_is_prefix() {
3534 assert_eq!(shared_prefix("foo", "foobar"), 3);
3535 }
3536
3537 #[test]
3540 fn detect_format_csv() {
3541 let path = std::path::Path::new("data.csv");
3542 assert_eq!(detect_format(path).unwrap(), DataFormat::Csv);
3543 }
3544
3545 #[test]
3546 fn detect_format_tsv() {
3547 assert_eq!(
3548 detect_format(std::path::Path::new("data.tsv")).unwrap(),
3549 DataFormat::Tsv
3550 );
3551 assert_eq!(
3552 detect_format(std::path::Path::new("data.txt")).unwrap(),
3553 DataFormat::Tsv
3554 );
3555 assert_eq!(
3556 detect_format(std::path::Path::new("data.tab")).unwrap(),
3557 DataFormat::Tsv
3558 );
3559 }
3560
3561 #[test]
3562 fn detect_format_parquet() {
3563 assert_eq!(
3564 detect_format(std::path::Path::new("data.parquet")).unwrap(),
3565 DataFormat::Parquet
3566 );
3567 assert_eq!(
3568 detect_format(std::path::Path::new("data.pq")).unwrap(),
3569 DataFormat::Parquet
3570 );
3571 assert_eq!(
3572 detect_format(std::path::Path::new("data.pqt")).unwrap(),
3573 DataFormat::Parquet
3574 );
3575 }
3576
3577 #[test]
3578 fn detect_format_uppercase_extension() {
3579 assert_eq!(
3580 detect_format(std::path::Path::new("data.CSV")).unwrap(),
3581 DataFormat::Csv
3582 );
3583 }
3584
3585 #[test]
3586 fn detect_format_unknown_extension_is_error() {
3587 let err = detect_format(std::path::Path::new("data.json")).unwrap_err();
3588 let msg = format!("{err:?}");
3589 assert!(
3590 msg.contains("json") || msg.contains("unsupported"),
3591 "error should mention extension, got: {msg}"
3592 );
3593 }
3594
3595 #[test]
3598 fn strip_categorical_sentinel_marked_cell() {
3599 let marked = "\u{0}hello";
3601 let (text, found) = strip_categorical_sentinel(marked);
3602 assert!(found);
3603 assert_eq!(text, "hello");
3604 }
3605
3606 #[test]
3607 fn strip_categorical_sentinel_unmarked_cell() {
3608 let (text, found) = strip_categorical_sentinel("plain");
3609 assert!(!found);
3610 assert_eq!(text, "plain");
3611 }
3612
3613 #[test]
3614 fn strip_categorical_sentinel_empty_string() {
3615 let (text, found) = strip_categorical_sentinel("");
3616 assert!(!found);
3617 assert_eq!(text, "");
3618 }
3619
3620 #[test]
3621 fn strip_categorical_sentinel_only_sentinel() {
3622 let s = "\u{0}";
3623 let (text, found) = strip_categorical_sentinel(s);
3624 assert!(found);
3625 assert_eq!(text, "");
3626 }
3627
3628 #[test]
3631 fn projected_headers_selects_by_index() {
3632 let all = vec![
3633 "a".to_string(),
3634 "b".to_string(),
3635 "c".to_string(),
3636 "d".to_string(),
3637 ];
3638 let selected = projected_headers(&all, &[1, 3]);
3639 assert_eq!(selected, vec!["b".to_string(), "d".to_string()]);
3640 }
3641
3642 #[test]
3643 fn projected_headers_empty_selection() {
3644 let all = vec!["x".to_string(), "y".to_string()];
3645 let selected = projected_headers(&all, &[]);
3646 assert!(selected.is_empty());
3647 }
3648
3649 #[test]
3650 fn projected_headers_all_indices() {
3651 let all = vec!["p".to_string(), "q".to_string()];
3652 let selected = projected_headers(&all, &[0, 1]);
3653 assert_eq!(selected, all);
3654 }
3655
3656 #[test]
3657 fn canonical_level_bits_collapses_signed_zero() {
3658 let pos = 0.0_f64;
3662 let neg = -0.0_f64;
3663 assert_ne!(
3664 pos.to_bits(),
3665 neg.to_bits(),
3666 "precondition: raw bits differ"
3667 );
3668 assert_eq!(pos, neg, "precondition: numerically equal");
3669 assert_eq!(canonical_level_bits(pos), canonical_level_bits(neg));
3670 assert_eq!(canonical_level_bits(neg), 0.0_f64.to_bits());
3671 assert_eq!(canonical_level_bits(-1.0 * 0.0), 0.0_f64.to_bits());
3673 assert_eq!(canonical_level_bits(0.0 - 0.0), 0.0_f64.to_bits());
3674 }
3675
3676 #[test]
3677 fn canonical_level_bits_is_bit_stable_on_ordinary_values() {
3678 for &v in &[
3681 1.0_f64,
3682 -1.0,
3683 2.5,
3684 -3.75,
3685 1e300,
3686 -1e-300,
3687 f64::MIN,
3688 f64::MAX,
3689 ] {
3690 assert_eq!(canonical_level_bits(v), v.to_bits(), "value {v}");
3691 }
3692 assert_ne!(canonical_level_bits(1.0), canonical_level_bits(2.0));
3694 assert_ne!(canonical_level_bits(0.0), canonical_level_bits(1.0));
3695 assert_ne!(
3697 canonical_level_bits(f64::INFINITY),
3698 canonical_level_bits(f64::NEG_INFINITY)
3699 );
3700 }
3701
3702 #[test]
3703 fn canonical_level_bits_collapses_nan_payloads() {
3704 let a = f64::NAN;
3706 let b = f64::from_bits(0x7ff8_0000_0000_0001); let c = -f64::NAN; assert!(a.is_nan() && b.is_nan() && c.is_nan());
3709 assert_eq!(canonical_level_bits(a), canonical_level_bits(b));
3710 assert_eq!(canonical_level_bits(a), canonical_level_bits(c));
3711 }
3712
3713 #[test]
3714 fn canonical_level_bits_is_idempotent() {
3715 for &v in &[0.0_f64, -0.0, 1.0, -2.0, f64::NAN] {
3718 let once = canonical_level_bits(v);
3719 let twice = canonical_level_bits(f64::from_bits(once));
3720 assert_eq!(once, twice, "value {v}");
3721 }
3722 }
3723}