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.get(selected_idx).unwrap().trim(),
591 total_rows,
592 &headers[j],
593 )?;
594 }
595 }
596
597 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
598 if stream_ms > 100.0 {
599 log::info!(
600 "[DATA-LOAD] delim_stream | n_rows={} | n_cols={} | {:.1}ms",
601 total_rows,
602 p,
603 stream_ms
604 );
605 }
606
607 if total_rows == 0 {
608 return Err(DataError::EmptyInput {
609 reason: "file has no rows".to_string(),
610 });
611 }
612
613 let t_schema = std::time::Instant::now();
614 let column_kinds = inference
615 .iter()
616 .enumerate()
617 .map(|(j, state)| state.kind(categorical_roles.contains(headers[j].as_str())))
618 .collect::<Vec<_>>();
619 let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
620 if schema_ms > 100.0 {
621 let n_cat = column_kinds
622 .iter()
623 .filter(|k| matches!(k, ColumnKindTag::Categorical))
624 .count();
625 log::info!(
626 "[DATA-LOAD] delim_convert+infer | n_cols={} | n_cat={} | {:.1}ms",
627 p,
628 n_cat,
629 schema_ms
630 );
631 }
632
633 let t_assemble = std::time::Instant::now();
638 let mut values = Array2::<f64>::zeros((total_rows, p));
639 let mut categorical_encoders = (0..p)
640 .map(|j| {
641 matches!(column_kinds[j], ColumnKindTag::Categorical).then(CategoricalEncoder::default)
642 })
643 .collect::<Vec<_>>();
644 let mut encode_rdr = ReaderBuilder::new()
645 .has_headers(true)
646 .delimiter(delimiter)
647 .from_path(path)
648 .map_err(|e| DataError::ParseError {
649 reason: format!("failed to reopen '{}': {e}", path.display()),
650 })?;
651 encode_rdr.headers().map_err(|e| DataError::ParseError {
652 reason: format!("failed to reread headers: {e}"),
653 })?;
654 let mut encoded_rows = 0usize;
655 while encode_rdr
656 .read_record(&mut record)
657 .map_err(|e| DataError::ParseError {
658 reason: format!("failed reading row: {e}"),
659 })?
660 {
661 if record.len() != all_headers.len() {
662 return Err(DataError::SchemaMismatch {
663 reason: format!(
664 "row width mismatch at row {}: got {} fields, expected {}",
665 encoded_rows + 1,
666 record.len(),
667 all_headers.len()
668 ),
669 });
670 }
671 if encoded_rows >= total_rows {
672 return Err(DataError::SchemaMismatch {
673 reason: "data file changed while its schema was being discovered".to_string(),
674 });
675 }
676 for (j, &selected_idx) in selected_indices.iter().enumerate() {
677 let raw = record.get(selected_idx).unwrap().trim();
678 values[[encoded_rows, j]] = match column_kinds[j] {
679 ColumnKindTag::Continuous | ColumnKindTag::Binary => {
680 parse_inferred_numeric_cell(raw, encoded_rows + 1, &headers[j])?
681 }
682 ColumnKindTag::Categorical => {
683 if raw.is_empty() {
684 return Err(DataError::EmptyInput {
685 reason: format!(
686 "empty field at row {}, column '{}'",
687 encoded_rows + 1,
688 &headers[j]
689 ),
690 });
691 }
692 categorical_encoders[j]
693 .as_mut()
694 .expect("categorical encoder")
695 .encode(raw) as f64
696 }
697 };
698 }
699 encoded_rows += 1;
700 }
701 if encoded_rows != total_rows {
702 return Err(DataError::SchemaMismatch {
703 reason: "data file changed while its schema was being discovered".to_string(),
704 });
705 }
706
707 let mut levels = vec![Vec::<String>::new(); p];
708 for (j, encoder) in categorical_encoders.into_iter().enumerate() {
709 if let Some(encoder) = encoder {
710 levels[j] = encoder.finish(values.column_mut(j), LevelOrder::Canonical);
711 }
712 }
713 let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
714 if assemble_ms > 100.0 {
715 log::info!(
716 "[DATA-LOAD] delim_assemble_array2 | n_rows={} | n_cols={} | {:.1}ms",
717 total_rows,
718 p,
719 assemble_ms
720 );
721 }
722
723 let schema = DataSchema {
724 columns: headers
725 .iter()
726 .enumerate()
727 .map(|(j, name)| SchemaColumn {
728 name: name.clone(),
729 kind: column_kinds[j],
730 levels: std::mem::take(&mut levels[j]),
731 })
732 .collect(),
733 };
734 Ok(EncodedDataset {
735 headers,
736 values,
737 schema,
738 column_kinds,
739 })
740}
741
742#[derive(Clone, Copy)]
743struct DelimitedInferenceState {
744 all_numeric: bool,
745 all_binary: bool,
746}
747
748impl Default for DelimitedInferenceState {
749 fn default() -> Self {
750 Self {
751 all_numeric: true,
752 all_binary: true,
753 }
754 }
755}
756
757impl DelimitedInferenceState {
758 fn observe(&mut self, raw: &str, row: usize, header: &str) -> Result<(), DataError> {
759 if raw.is_empty() {
760 return Err(DataError::EmptyInput {
761 reason: format!("empty field at row {row}, column '{header}'"),
762 });
763 }
764 match raw.parse::<f64>() {
765 Ok(value) => {
766 if !value.is_finite() {
767 return Err(DataError::InvalidValue {
768 reason: format!("non-finite value at row {row}, column '{header}'"),
769 });
770 }
771 if (value - 0.0).abs() >= 1e-12 && (value - 1.0).abs() >= 1e-12 {
772 self.all_binary = false;
773 }
774 }
775 Err(_) => {
776 self.all_numeric = false;
777 self.all_binary = false;
778 }
779 }
780 Ok(())
781 }
782
783 fn kind(self, force_categorical: bool) -> ColumnKindTag {
784 if force_categorical || !self.all_numeric {
785 ColumnKindTag::Categorical
786 } else if self.all_binary {
787 ColumnKindTag::Binary
788 } else {
789 ColumnKindTag::Continuous
790 }
791 }
792}
793
794fn parse_inferred_numeric_cell(raw: &str, row: usize, header: &str) -> Result<f64, DataError> {
795 if raw.is_empty() {
796 return Err(DataError::EmptyInput {
797 reason: format!("empty field at row {row}, column '{header}'"),
798 });
799 }
800 let value = raw
801 .parse::<f64>()
802 .map_err(|error| DataError::EncodingFailure {
803 reason: format!(
804 "failed to parse numeric value '{raw}' at row {row}, column '{header}': {error}"
805 ),
806 })?;
807 if !value.is_finite() {
808 return Err(DataError::InvalidValue {
809 reason: format!("non-finite value at row {row}, column '{header}'"),
810 });
811 }
812 Ok(value)
813}
814
815#[derive(Clone, Copy)]
816enum LevelOrder {
817 Encounter,
818 Canonical,
819}
820
821#[derive(Default)]
830struct CategoricalEncoder {
831 encounter_codes: HashMap<String, usize>,
832}
833
834impl CategoricalEncoder {
835 fn encode(&mut self, label: &str) -> usize {
836 if let Some(&code) = self.encounter_codes.get(label) {
837 return code;
838 }
839 let code = self.encounter_codes.len();
840 self.encounter_codes.insert(label.to_owned(), code);
841 code
842 }
843
844 fn finish(self, mut encoded: ArrayViewMut1<'_, f64>, order: LevelOrder) -> Vec<String> {
845 match order {
846 LevelOrder::Encounter => {
847 let mut levels = std::iter::repeat_with(|| None)
848 .take(self.encounter_codes.len())
849 .collect::<Vec<Option<String>>>();
850 for (level, old_code) in self.encounter_codes {
851 levels[old_code] = Some(level);
852 }
853 levels
854 .into_iter()
855 .map(|level| level.expect("encounter code must name one level"))
856 .collect()
857 }
858 LevelOrder::Canonical => {
859 let mut levels_with_old_codes =
860 self.encounter_codes.into_iter().collect::<Vec<_>>();
861 levels_with_old_codes
862 .sort_by(|(a, _), (b, _)| natural_level_cmp(a.as_str(), b.as_str()));
863 let mut remap = vec![0usize; levels_with_old_codes.len()];
864 for (new_code, (_, old_code)) in levels_with_old_codes.iter().enumerate() {
865 remap[*old_code] = new_code;
866 }
867 for code in encoded.iter_mut() {
868 *code = remap[*code as usize] as f64;
869 }
870 levels_with_old_codes
871 .into_iter()
872 .map(|(level, _)| level)
873 .collect()
874 }
875 }
876 }
877}
878
879fn load_delimited_with_schema(
880 path: &Path,
881 delimiter: u8,
882 schema: &DataSchema,
883 unseen_policy: UnseenCategoryPolicy,
884 requested_columns: &[String],
885) -> Result<EncodedDataset, DataError> {
886 let t_open = std::time::Instant::now();
887 let mut rdr = ReaderBuilder::new()
888 .has_headers(true)
889 .delimiter(delimiter)
890 .from_path(path)
891 .map_err(|e| DataError::ParseError {
892 reason: format!("failed to open '{}': {e}", path.display()),
893 })?;
894
895 let all_headers: Vec<String> = rdr
896 .headers()
897 .map_err(|e| DataError::ParseError {
898 reason: format!("failed to read headers: {e}"),
899 })?
900 .iter()
901 .map(|s| s.trim().to_string())
902 .collect();
903 if all_headers.is_empty() {
904 return Err(DataError::EmptyInput {
905 reason: "file has no headers".to_string(),
906 });
907 }
908 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
909 let headers = projected_headers(&all_headers, &selected_indices);
910 let p = headers.len();
911 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
912 if open_ms > 100.0 {
913 log::info!(
914 "[DATA-LOAD] delim_schema_open+headers | n_headers={} | n_proj={} | {:.1}ms",
915 all_headers.len(),
916 p,
917 open_ms
918 );
919 }
920
921 let schema_byname: HashMap<&str, &SchemaColumn> = schema
923 .columns
924 .iter()
925 .map(|c| (c.name.as_str(), c))
926 .collect();
927
928 let mut col_meta = Vec::<ColMeta>::with_capacity(p);
929 for name in &headers {
930 if let Some(sc) = schema_byname.get(name.as_str()) {
931 let level_map = if matches!(sc.kind, ColumnKindTag::Categorical) {
932 Some(
933 sc.levels
934 .iter()
935 .enumerate()
936 .map(|(idx, v)| (v.as_str(), idx as f64))
937 .collect::<HashMap<_, _>>(),
938 )
939 } else {
940 None
941 };
942 col_meta.push(ColMeta {
943 kind: sc.kind,
944 level_map,
945 schema_col: (*sc).clone(),
946 });
947 } else {
948 col_meta.push(ColMeta {
950 kind: ColumnKindTag::Continuous, level_map: None,
952 schema_col: SchemaColumn {
953 name: name.clone(),
954 kind: ColumnKindTag::Continuous,
955 levels: Vec::new(),
956 },
957 });
958 }
959 }
960
961 let needs_inference: Vec<bool> = headers
963 .iter()
964 .map(|h| !schema_byname.contains_key(h.as_str()))
965 .collect();
966
967 if needs_inference.iter().all(|needs| !needs) {
973 let t_stream = std::time::Instant::now();
974 let mut flat_values = Vec::<f64>::new();
975 let mut total_rows = 0usize;
976 let mut record = StringRecord::new();
977 while rdr
978 .read_record(&mut record)
979 .map_err(|e| DataError::ParseError {
980 reason: format!("failed reading row: {e}"),
981 })?
982 {
983 if record.len() != all_headers.len() {
984 return Err(DataError::SchemaMismatch {
985 reason: format!(
986 "row width mismatch at row {}: got {} fields, expected {}",
987 total_rows + 1,
988 record.len(),
989 all_headers.len()
990 ),
991 });
992 }
993 total_rows += 1;
994 for j in 0..p {
995 let raw = record.get(selected_indices[j]).unwrap().trim();
996 flat_values.push(parse_cell_with_schema(
997 raw,
998 &col_meta[j],
999 total_rows,
1000 &headers[j],
1001 &unseen_policy,
1002 )?);
1003 }
1004 }
1005 if total_rows == 0 {
1006 return Err(DataError::EmptyInput {
1007 reason: "file has no rows".to_string(),
1008 });
1009 }
1010 let values = Array2::from_shape_vec((total_rows, p), flat_values).map_err(|error| {
1011 DataError::EncodingFailure {
1012 reason: format!("failed to assemble schema-guided delimited matrix: {error}"),
1013 }
1014 })?;
1015 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
1016 if stream_ms > 100.0 {
1017 log::info!(
1018 "[DATA-LOAD] delim_schema_direct | n_rows={} | n_cols={} | {:.1}ms",
1019 total_rows,
1020 p,
1021 stream_ms
1022 );
1023 }
1024 let column_kinds = col_meta.iter().map(|meta| meta.kind).collect();
1025 let schema_out = DataSchema {
1026 columns: col_meta.into_iter().map(|meta| meta.schema_col).collect(),
1027 };
1028 return Ok(EncodedDataset {
1029 headers,
1030 values,
1031 schema: schema_out,
1032 column_kinds,
1033 });
1034 }
1035
1036 let mut inference = vec![DelimitedInferenceState::default(); p];
1039 let mut total_rows: usize = 0;
1040 let t_stream = std::time::Instant::now();
1041 let mut record = StringRecord::new();
1042 while rdr
1043 .read_record(&mut record)
1044 .map_err(|e| DataError::ParseError {
1045 reason: format!("failed reading row: {e}"),
1046 })?
1047 {
1048 if record.len() != all_headers.len() {
1049 return Err(DataError::SchemaMismatch {
1050 reason: format!(
1051 "row width mismatch at row {}: got {} fields, expected {}",
1052 total_rows + 1,
1053 record.len(),
1054 all_headers.len()
1055 ),
1056 });
1057 }
1058 total_rows += 1;
1059
1060 for j in 0..p {
1061 let raw = record.get(selected_indices[j]).unwrap().trim();
1062 if needs_inference[j] {
1063 inference[j].observe(raw, total_rows, &headers[j])?;
1064 } else {
1065 parse_cell_with_schema(raw, &col_meta[j], total_rows, &headers[j], &unseen_policy)?;
1066 }
1067 }
1068 }
1069
1070 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
1071 if stream_ms > 100.0 {
1072 let n_inf = needs_inference.iter().filter(|x| **x).count();
1073 log::info!(
1074 "[DATA-LOAD] delim_schema_stream | n_rows={} | n_cols={} | n_inf={} | {:.1}ms",
1075 total_rows,
1076 p,
1077 n_inf,
1078 stream_ms
1079 );
1080 }
1081
1082 if total_rows == 0 {
1083 return Err(DataError::EmptyInput {
1084 reason: "file has no rows".to_string(),
1085 });
1086 }
1087
1088 let t_finalize = std::time::Instant::now();
1089 for j in 0..p {
1090 if needs_inference[j] {
1091 let kind = inference[j].kind(false);
1092 col_meta[j].kind = kind;
1093 col_meta[j].schema_col.kind = kind;
1094 }
1095 }
1096 let finalize_ms = t_finalize.elapsed().as_secs_f64() * 1000.0;
1097 if finalize_ms > 100.0 {
1098 log::info!(
1099 "[DATA-LOAD] delim_schema_finalize | n_cols={} | {:.1}ms",
1100 p,
1101 finalize_ms
1102 );
1103 }
1104
1105 let t_assemble = std::time::Instant::now();
1109 let mut values = Array2::<f64>::zeros((total_rows, p));
1110 let mut inferred_encoders = (0..p)
1111 .map(|j| {
1112 (needs_inference[j] && matches!(col_meta[j].kind, ColumnKindTag::Categorical))
1113 .then(CategoricalEncoder::default)
1114 })
1115 .collect::<Vec<_>>();
1116 let mut encode_rdr = ReaderBuilder::new()
1117 .has_headers(true)
1118 .delimiter(delimiter)
1119 .from_path(path)
1120 .map_err(|e| DataError::ParseError {
1121 reason: format!("failed to reopen '{}': {e}", path.display()),
1122 })?;
1123 encode_rdr.headers().map_err(|e| DataError::ParseError {
1124 reason: format!("failed to reread headers: {e}"),
1125 })?;
1126 let mut encoded_rows = 0usize;
1127 while encode_rdr
1128 .read_record(&mut record)
1129 .map_err(|e| DataError::ParseError {
1130 reason: format!("failed reading row: {e}"),
1131 })?
1132 {
1133 if record.len() != all_headers.len() {
1134 return Err(DataError::SchemaMismatch {
1135 reason: format!(
1136 "row width mismatch at row {}: got {} fields, expected {}",
1137 encoded_rows + 1,
1138 record.len(),
1139 all_headers.len()
1140 ),
1141 });
1142 }
1143 if encoded_rows >= total_rows {
1144 return Err(DataError::SchemaMismatch {
1145 reason: "data file changed while its schema was being discovered".to_string(),
1146 });
1147 }
1148 for j in 0..p {
1149 let raw = record.get(selected_indices[j]).unwrap().trim();
1150 values[[encoded_rows, j]] = if !needs_inference[j] {
1151 parse_cell_with_schema(
1152 raw,
1153 &col_meta[j],
1154 encoded_rows + 1,
1155 &headers[j],
1156 &unseen_policy,
1157 )?
1158 } else {
1159 match col_meta[j].kind {
1160 ColumnKindTag::Continuous | ColumnKindTag::Binary => {
1161 parse_inferred_numeric_cell(raw, encoded_rows + 1, &headers[j])?
1162 }
1163 ColumnKindTag::Categorical => {
1164 if raw.is_empty() {
1165 return Err(DataError::EmptyInput {
1166 reason: format!(
1167 "empty field at row {}, column '{}'",
1168 encoded_rows + 1,
1169 &headers[j]
1170 ),
1171 });
1172 }
1173 let encoder = inferred_encoders[j]
1174 .as_mut()
1175 .expect("inferred categorical encoder");
1176 encoder.encode(raw) as f64
1177 }
1178 }
1179 };
1180 }
1181 encoded_rows += 1;
1182 }
1183 if encoded_rows != total_rows {
1184 return Err(DataError::SchemaMismatch {
1185 reason: "data file changed while its schema was being discovered".to_string(),
1186 });
1187 }
1188 for (j, encoder) in inferred_encoders.into_iter().enumerate() {
1189 if let Some(encoder) = encoder {
1190 col_meta[j].schema_col.levels =
1191 encoder.finish(values.column_mut(j), LevelOrder::Canonical);
1192 }
1193 }
1194 let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
1195 if assemble_ms > 100.0 {
1196 log::info!(
1197 "[DATA-LOAD] delim_schema_assemble | n_rows={} | n_cols={} | {:.1}ms",
1198 total_rows,
1199 p,
1200 assemble_ms
1201 );
1202 }
1203
1204 let column_kinds = col_meta.iter().map(|meta| meta.kind).collect();
1205 let schema_out = DataSchema {
1206 columns: col_meta.into_iter().map(|m| m.schema_col).collect(),
1207 };
1208 Ok(EncodedDataset {
1209 headers,
1210 values,
1211 schema: schema_out,
1212 column_kinds,
1213 })
1214}
1215
1216fn parse_cell_with_schema(
1217 raw: &str,
1218 meta: &ColMeta<'_>,
1219 row: usize,
1220 col_name: &str,
1221 unseen_policy: &UnseenCategoryPolicy,
1222) -> Result<f64, DataError> {
1223 let val = match meta.kind {
1224 ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
1225 DataError::SchemaMismatch {
1226 reason: format!(
1227 "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
1228 col_name, row, raw, err
1229 ),
1230 }
1231 })?,
1232 ColumnKindTag::Binary => {
1233 let v = raw
1234 .parse::<f64>()
1235 .map_err(|err| DataError::SchemaMismatch {
1236 reason: format!(
1237 "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
1238 col_name, row, raw, err
1239 ),
1240 })?;
1241 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
1242 return Err(DataError::SchemaMismatch {
1243 reason: format!(
1244 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
1245 col_name, row, v
1246 ),
1247 });
1248 }
1249 v
1250 }
1251 ColumnKindTag::Categorical => {
1252 let map = meta
1253 .level_map
1254 .as_ref()
1255 .ok_or_else(|| DataError::EncodingFailure {
1256 reason: "internal categorical schema map missing".to_string(),
1257 })?;
1258 match map.get(raw) {
1259 Some(v) => *v,
1260 None => unseen_policy
1261 .unseen_code_for(col_name, meta.schema_col.levels.len())
1262 .ok_or_else(|| DataError::SchemaMismatch {
1263 reason: format!(
1264 "unseen level '{}' in categorical column '{}' at row {}",
1265 raw, col_name, row
1266 ),
1267 })?,
1268 }
1269 }
1270 };
1271 if !val.is_finite() {
1272 return Err(DataError::InvalidValue {
1273 reason: format!("non-finite value at row {}, column '{}'", row, col_name),
1274 });
1275 }
1276 Ok(val)
1277}
1278
1279struct ColMeta<'a> {
1282 kind: ColumnKindTag,
1283 level_map: Option<HashMap<&'a str, f64>>,
1284 schema_col: SchemaColumn,
1285}
1286
1287fn arrow_field_is_string(dt: &arrow::datatypes::DataType) -> bool {
1300 use arrow::datatypes::DataType;
1301 match dt {
1302 DataType::Utf8 | DataType::LargeUtf8 => true,
1303 DataType::Dictionary(_, value_type) => arrow_field_is_string(value_type),
1304 _ => false,
1305 }
1306}
1307
1308fn write_arrow_numeric_values(
1309 values: impl IntoIterator<Item = f64>,
1310 base_row: usize,
1311 header: &str,
1312 mut output: ArrayViewMut1<'_, f64>,
1313 categorical_encoder: Option<&mut CategoricalEncoder>,
1314 all_binary: &mut bool,
1315) -> Result<(), DataError> {
1316 match categorical_encoder {
1317 Some(encoder) => {
1318 for (batch_row, value) in values.into_iter().enumerate() {
1319 if !value.is_finite() {
1320 return Err(DataError::InvalidValue {
1321 reason: format!(
1322 "non-finite value at row {}, column '{}'",
1323 base_row + batch_row + 1,
1324 header
1325 ),
1326 });
1327 }
1328 output[batch_row] = encoder.encode(&value.to_string()) as f64;
1329 }
1330 }
1331 None => {
1332 for (batch_row, value) in values.into_iter().enumerate() {
1333 if !value.is_finite() {
1334 return Err(DataError::InvalidValue {
1335 reason: format!(
1336 "non-finite value at row {}, column '{}'",
1337 base_row + batch_row + 1,
1338 header
1339 ),
1340 });
1341 }
1342 if (value - 0.0).abs() >= 1e-12 && (value - 1.0).abs() >= 1e-12 {
1343 *all_binary = false;
1344 }
1345 output[batch_row] = value;
1346 }
1347 }
1348 }
1349 Ok(())
1350}
1351
1352fn reject_arrow_null_values(
1353 col: &dyn arrow::array::Array,
1354 base_row: usize,
1355 header: &str,
1356) -> Result<(), DataError> {
1357 let Some(nulls) = col.logical_nulls() else {
1358 return Ok(());
1359 };
1360 if let Some(batch_row) = (0..col.len()).find(|&row| nulls.is_null(row)) {
1361 return Err(DataError::InvalidValue {
1362 reason: format!(
1363 "null value at row {}, column '{}'",
1364 base_row + batch_row + 1,
1365 header
1366 ),
1367 });
1368 }
1369 Ok(())
1370}
1371
1372fn arrow_dictionary_string_value_at<'a, K>(
1373 col: &'a dyn arrow::array::Array,
1374 index: usize,
1375 logical_row: usize,
1376 header: &str,
1377) -> Result<&'a str, DataError>
1378where
1379 K: arrow::datatypes::ArrowDictionaryKeyType,
1380{
1381 use arrow::array::DictionaryArray;
1382
1383 let dictionary = col
1384 .as_any()
1385 .downcast_ref::<DictionaryArray<K>>()
1386 .ok_or_else(|| DataError::EncodingFailure {
1387 reason: format!(
1388 "Arrow dictionary column '{}' did not match its declared key type",
1389 header
1390 ),
1391 })?;
1392 let value_index = dictionary
1393 .key(index)
1394 .ok_or_else(|| DataError::InvalidValue {
1395 reason: format!("null value at row {logical_row}, column '{header}'"),
1396 })?;
1397 if value_index >= dictionary.values().len() {
1398 return Err(DataError::EncodingFailure {
1399 reason: format!(
1400 "Arrow dictionary column '{}' has out-of-range key {} at row {}",
1401 header, value_index, logical_row
1402 ),
1403 });
1404 }
1405 arrow_string_value_at(
1406 dictionary.values().as_ref(),
1407 value_index,
1408 logical_row,
1409 header,
1410 )
1411}
1412
1413fn arrow_string_value_at<'a>(
1416 col: &'a dyn arrow::array::Array,
1417 index: usize,
1418 logical_row: usize,
1419 header: &str,
1420) -> Result<&'a str, DataError> {
1421 use arrow::array::{LargeStringArray, StringArray};
1422 use arrow::datatypes::{
1423 DataType, Int8Type, Int16Type, Int32Type, Int64Type, UInt8Type, UInt16Type, UInt32Type,
1424 UInt64Type,
1425 };
1426
1427 if index >= col.len() {
1428 return Err(DataError::EncodingFailure {
1429 reason: format!(
1430 "Arrow string column '{}' has out-of-range index {} at row {}",
1431 header, index, logical_row
1432 ),
1433 });
1434 }
1435 if col.is_null(index) {
1436 return Err(DataError::InvalidValue {
1437 reason: format!("null value at row {logical_row}, column '{header}'"),
1438 });
1439 }
1440
1441 match col.data_type() {
1442 DataType::Utf8 => col
1443 .as_any()
1444 .downcast_ref::<StringArray>()
1445 .map(|array| array.value(index))
1446 .ok_or_else(|| DataError::EncodingFailure {
1447 reason: format!("Arrow column '{}' could not be read as Utf8", header),
1448 }),
1449 DataType::LargeUtf8 => col
1450 .as_any()
1451 .downcast_ref::<LargeStringArray>()
1452 .map(|array| array.value(index))
1453 .ok_or_else(|| DataError::EncodingFailure {
1454 reason: format!("Arrow column '{}' could not be read as LargeUtf8", header),
1455 }),
1456 DataType::Dictionary(key_type, _) => match key_type.as_ref() {
1457 DataType::Int8 => {
1458 arrow_dictionary_string_value_at::<Int8Type>(col, index, logical_row, header)
1459 }
1460 DataType::Int16 => {
1461 arrow_dictionary_string_value_at::<Int16Type>(col, index, logical_row, header)
1462 }
1463 DataType::Int32 => {
1464 arrow_dictionary_string_value_at::<Int32Type>(col, index, logical_row, header)
1465 }
1466 DataType::Int64 => {
1467 arrow_dictionary_string_value_at::<Int64Type>(col, index, logical_row, header)
1468 }
1469 DataType::UInt8 => {
1470 arrow_dictionary_string_value_at::<UInt8Type>(col, index, logical_row, header)
1471 }
1472 DataType::UInt16 => {
1473 arrow_dictionary_string_value_at::<UInt16Type>(col, index, logical_row, header)
1474 }
1475 DataType::UInt32 => {
1476 arrow_dictionary_string_value_at::<UInt32Type>(col, index, logical_row, header)
1477 }
1478 DataType::UInt64 => {
1479 arrow_dictionary_string_value_at::<UInt64Type>(col, index, logical_row, header)
1480 }
1481 other => Err(DataError::InvalidValue {
1482 reason: format!(
1483 "unsupported Arrow dictionary key type {:?} for column '{}'",
1484 other, header
1485 ),
1486 }),
1487 },
1488 other => Err(DataError::InvalidValue {
1489 reason: format!(
1490 "unsupported Arrow string column type {:?} for column '{}'",
1491 other, header
1492 ),
1493 }),
1494 }
1495}
1496
1497fn decode_arrow_batch_column_into(
1504 col: &dyn arrow::array::Array,
1505 base_row: usize,
1506 header: &str,
1507 is_string_col: bool,
1508 mut output: ArrayViewMut1<'_, f64>,
1509 mut categorical_encoder: Option<&mut CategoricalEncoder>,
1510 all_binary: &mut bool,
1511) -> Result<(), DataError> {
1512 use arrow::array::{
1513 BooleanArray, Float32Array, Float64Array, Int8Array, Int16Array, Int32Array, Int64Array,
1514 UInt8Array, UInt16Array, UInt32Array, UInt64Array,
1515 };
1516 use arrow::datatypes::DataType;
1517
1518 let n_rows = output.len();
1519 if col.len() != n_rows {
1520 return Err(DataError::SchemaMismatch {
1521 reason: format!(
1522 "Arrow column '{}' has {} rows, but its record batch has {}",
1523 header,
1524 col.len(),
1525 n_rows
1526 ),
1527 });
1528 }
1529 reject_arrow_null_values(col, base_row, header)?;
1530
1531 if is_string_col {
1532 let encoder =
1533 categorical_encoder
1534 .as_deref_mut()
1535 .ok_or_else(|| DataError::EncodingFailure {
1536 reason: format!("categorical Arrow encoder missing for column '{header}'"),
1537 })?;
1538 for batch_row in 0..n_rows {
1539 let label = arrow_string_value_at(col, batch_row, base_row + batch_row + 1, header)?;
1540 output[batch_row] = encoder.encode(label) as f64;
1541 }
1542 return Ok(());
1543 }
1544
1545 let decoded_col;
1552 let col: &dyn arrow::array::Array = if let DataType::Dictionary(_, value_type) = col.data_type()
1553 {
1554 decoded_col = arrow::compute::cast(col, value_type).map_err(|e| DataError::ParseError {
1555 reason: format!(
1556 "failed to decode dictionary-encoded numeric column '{}': {e}",
1557 header
1558 ),
1559 })?;
1560 decoded_col.as_ref()
1561 } else {
1562 col
1563 };
1564 reject_arrow_null_values(col, base_row, header)?;
1567
1568 macro_rules! write_primitive {
1569 ($array_type:ty, $convert:expr) => {{
1570 let array = col.as_any().downcast_ref::<$array_type>().unwrap();
1571 write_arrow_numeric_values(
1572 array.values().iter().copied().map($convert),
1573 base_row,
1574 header,
1575 output,
1576 categorical_encoder,
1577 all_binary,
1578 )
1579 }};
1580 }
1581
1582 match col.data_type() {
1583 DataType::Float64 => write_primitive!(Float64Array, |value: f64| value),
1584 DataType::Float32 => write_primitive!(Float32Array, |value: f32| value as f64),
1585 DataType::Int64 => write_primitive!(Int64Array, |value: i64| value as f64),
1586 DataType::Int32 => write_primitive!(Int32Array, |value: i32| value as f64),
1587 DataType::Int16 => write_primitive!(Int16Array, |value: i16| value as f64),
1588 DataType::Int8 => write_primitive!(Int8Array, |value: i8| value as f64),
1589 DataType::UInt64 => write_primitive!(UInt64Array, |value: u64| value as f64),
1590 DataType::UInt32 => write_primitive!(UInt32Array, |value: u32| value as f64),
1591 DataType::UInt16 => write_primitive!(UInt16Array, |value: u16| value as f64),
1592 DataType::UInt8 => write_primitive!(UInt8Array, |value: u8| value as f64),
1593 DataType::Boolean => {
1594 let arr = col.as_any().downcast_ref::<BooleanArray>().unwrap();
1595 write_arrow_numeric_values(
1596 (0..n_rows).map(|i| if arr.value(i) { 1.0 } else { 0.0 }),
1597 base_row,
1598 header,
1599 output,
1600 categorical_encoder,
1601 all_binary,
1602 )
1603 }
1604 other => Err(DataError::InvalidValue {
1605 reason: format!(
1606 "unsupported Arrow column type {:?} for column '{}'",
1607 other, header
1608 ),
1609 }),
1610 }
1611}
1612
1613pub fn encode_arrow_record_batch_reader_with_inferred_schema(
1627 reader: &mut dyn arrow::record_batch::RecordBatchReader,
1628 headers: Vec<String>,
1629) -> Result<EncodedDataset, DataError> {
1630 if headers.is_empty() {
1631 return Err(DataError::EmptyInput {
1632 reason: "Arrow table must have at least one header column".to_string(),
1633 });
1634 }
1635
1636 let mut seen_headers = HashSet::<&str>::with_capacity(headers.len());
1637 for (column, header) in headers.iter().enumerate() {
1638 if header.trim().is_empty() {
1639 return Err(DataError::EmptyInput {
1640 reason: format!("Arrow header at column {} cannot be empty", column + 1),
1641 });
1642 }
1643 if !seen_headers.insert(header.as_str()) {
1644 return Err(DataError::SchemaMismatch {
1645 reason: format!("duplicate Arrow header '{}'", header),
1646 });
1647 }
1648 }
1649
1650 let arrow_schema = reader.schema();
1651 let p = headers.len();
1652 if arrow_schema.fields().len() != p {
1653 return Err(DataError::SchemaMismatch {
1654 reason: format!(
1655 "Arrow schema has {} columns, but {} normalized headers were supplied",
1656 arrow_schema.fields().len(),
1657 p
1658 ),
1659 });
1660 }
1661
1662 let is_string_col = arrow_schema
1663 .fields()
1664 .iter()
1665 .map(|field| arrow_field_is_string(field.data_type()))
1666 .collect::<Vec<_>>();
1667 let mut all_binary = vec![true; p];
1668 let mut categorical_encoders = is_string_col
1669 .iter()
1670 .map(|&is_string| is_string.then(CategoricalEncoder::default))
1671 .collect::<Vec<_>>();
1672 let mut encoded_values = Vec::<f64>::new();
1673 let mut rows_seen = 0usize;
1674
1675 for batch_result in reader {
1676 let batch = batch_result.map_err(|error| DataError::ParseError {
1677 reason: format!("failed to read Arrow record batch: {error}"),
1678 })?;
1679 if batch.num_columns() != p {
1680 return Err(DataError::SchemaMismatch {
1681 reason: format!(
1682 "Arrow record batch has {} columns, but {} normalized headers were supplied",
1683 batch.num_columns(),
1684 p
1685 ),
1686 });
1687 }
1688 for j in 0..p {
1689 let expected = arrow_schema.field(j).data_type();
1690 let actual = batch.column(j).data_type();
1691 if actual != expected {
1692 return Err(DataError::SchemaMismatch {
1693 reason: format!(
1694 "Arrow column '{}' changed type between schema and batch: expected {:?}, got {:?}",
1695 headers[j], expected, actual
1696 ),
1697 });
1698 }
1699 }
1700
1701 let n_rows = batch.num_rows();
1702 let batch_values = n_rows
1703 .checked_mul(p)
1704 .ok_or_else(|| DataError::EncodingFailure {
1705 reason: "Arrow batch dimensions do not fit in memory address space".to_string(),
1706 })?;
1707 let next_len = encoded_values
1708 .len()
1709 .checked_add(batch_values)
1710 .ok_or_else(|| DataError::EncodingFailure {
1711 reason: "Arrow dataset dimensions do not fit in memory address space".to_string(),
1712 })?;
1713 encoded_values
1714 .try_reserve(batch_values)
1715 .map_err(|error| DataError::EncodingFailure {
1716 reason: format!("failed to reserve Arrow dataset storage: {error}"),
1717 })?;
1718 let batch_offset = encoded_values.len();
1719 encoded_values.resize(next_len, 0.0);
1720 let mut batch_output = ndarray::ArrayViewMut2::from_shape(
1721 (n_rows, p),
1722 &mut encoded_values[batch_offset..next_len],
1723 )
1724 .map_err(|error| DataError::EncodingFailure {
1725 reason: format!("failed to shape Arrow batch output: {error}"),
1726 })?;
1727
1728 let decoded_columns = batch_output
1729 .axis_iter_mut(Axis(1))
1730 .into_par_iter()
1731 .zip(categorical_encoders.par_iter_mut())
1732 .zip(all_binary.par_iter_mut())
1733 .enumerate()
1734 .map(|(j, ((output, encoder), column_all_binary))| {
1735 decode_arrow_batch_column_into(
1736 batch.column(j).as_ref(),
1737 rows_seen,
1738 &headers[j],
1739 is_string_col[j],
1740 output,
1741 encoder.as_mut(),
1742 column_all_binary,
1743 )
1744 })
1745 .collect::<Vec<_>>();
1746 for decoded in decoded_columns {
1747 decoded?;
1748 }
1749 rows_seen = rows_seen
1750 .checked_add(n_rows)
1751 .ok_or_else(|| DataError::EncodingFailure {
1752 reason: "Arrow row count does not fit in memory address space".to_string(),
1753 })?;
1754 }
1755
1756 if rows_seen == 0 {
1757 return Err(DataError::EmptyInput {
1758 reason: "Arrow table data cannot be empty".to_string(),
1759 });
1760 }
1761
1762 let mut values = Array2::from_shape_vec((rows_seen, p), encoded_values).map_err(|error| {
1763 DataError::EncodingFailure {
1764 reason: format!("failed to shape encoded Arrow dataset: {error}"),
1765 }
1766 })?;
1767 let mut levels = vec![Vec::<String>::new(); p];
1768 for (j, encoder) in categorical_encoders.into_iter().enumerate() {
1769 if let Some(encoder) = encoder {
1770 levels[j] = encoder.finish(values.column_mut(j), LevelOrder::Canonical);
1771 }
1772 }
1773
1774 let mut schema_columns = Vec::<SchemaColumn>::with_capacity(p);
1775 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
1776 for (j, name) in headers.iter().enumerate() {
1777 let kind = if is_string_col[j] {
1778 ColumnKindTag::Categorical
1779 } else if all_binary[j] {
1780 ColumnKindTag::Binary
1781 } else {
1782 ColumnKindTag::Continuous
1783 };
1784 column_kinds.push(kind);
1785 schema_columns.push(SchemaColumn {
1786 name: name.clone(),
1787 kind,
1788 levels: std::mem::take(&mut levels[j]),
1789 });
1790 }
1791
1792 Ok(EncodedDataset {
1793 headers,
1794 values,
1795 schema: DataSchema {
1796 columns: schema_columns,
1797 },
1798 column_kinds,
1799 })
1800}
1801
1802fn load_parquet_inferred(
1803 path: &Path,
1804 requested_columns: &[String],
1805 categorical_roles: &HashSet<&str>,
1806) -> Result<EncodedDataset, DataError> {
1807 use parquet::arrow::{ProjectionMask, arrow_reader::ParquetRecordBatchReaderBuilder};
1808 use rayon::prelude::*;
1809 use std::fs::File;
1810
1811 let t_open = std::time::Instant::now();
1812 let file = File::open(path).map_err(|e| DataError::ParseError {
1813 reason: format!("failed to open parquet '{}': {e}", path.display()),
1814 })?;
1815 let builder =
1816 ParquetRecordBatchReaderBuilder::try_new(file).map_err(|e| DataError::ParseError {
1817 reason: format!("failed to read parquet metadata '{}': {e}", path.display()),
1818 })?;
1819
1820 let full_schema = builder.schema().clone();
1821 let all_headers: Vec<String> = full_schema
1822 .fields()
1823 .iter()
1824 .map(|f| f.name().clone())
1825 .collect();
1826 if all_headers.is_empty() {
1827 return Err(DataError::EmptyInput {
1828 reason: "parquet file has no columns".to_string(),
1829 });
1830 }
1831 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
1832 let headers = projected_headers(&all_headers, &selected_indices);
1833 let selected_fields = selected_indices
1834 .iter()
1835 .map(|&idx| full_schema.fields()[idx].clone())
1836 .collect::<Vec<_>>();
1837 let total_rows =
1838 usize::try_from(builder.metadata().file_metadata().num_rows()).map_err(|_| {
1839 DataError::ParseError {
1840 reason: "parquet row count does not fit in memory address space".to_string(),
1841 }
1842 })?;
1843 if total_rows == 0 {
1844 return Err(DataError::EmptyInput {
1845 reason: "parquet file has no rows".to_string(),
1846 });
1847 }
1848 let projection =
1849 ProjectionMask::roots(builder.parquet_schema(), selected_indices.iter().copied());
1850 let reader =
1851 builder
1852 .with_projection(projection)
1853 .build()
1854 .map_err(|e| DataError::ParseError {
1855 reason: format!("failed to build parquet reader: {e}"),
1856 })?;
1857 let p = headers.len();
1858 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
1859 if open_ms > 100.0 {
1860 log::info!(
1861 "[DATA-LOAD] parquet_open+meta | n_headers={} | n_proj={} | {:.1}ms",
1862 all_headers.len(),
1863 p,
1864 open_ms
1865 );
1866 }
1867
1868 let t_batches = std::time::Instant::now();
1869 let is_string_col = selected_fields
1870 .iter()
1871 .map(|field| arrow_field_is_string(field.data_type()))
1872 .collect::<Vec<_>>();
1873 let forced_numeric_categorical = headers
1874 .iter()
1875 .enumerate()
1876 .map(|(j, header)| !is_string_col[j] && categorical_roles.contains(header.as_str()))
1877 .collect::<Vec<_>>();
1878 let mut values = Array2::<f64>::zeros((total_rows, p));
1879 let mut all_binary = vec![true; p];
1880 let mut categorical_encoders = (0..p)
1881 .map(|j| {
1882 (is_string_col[j] || forced_numeric_categorical[j]).then(CategoricalEncoder::default)
1883 })
1884 .collect::<Vec<_>>();
1885 let mut rows_seen = 0usize;
1886 for batch_result in reader {
1887 let batch = batch_result.map_err(|e| DataError::ParseError {
1888 reason: format!("failed to read parquet record batch: {e}"),
1889 })?;
1890 let n_rows = batch.num_rows();
1891 if rows_seen.saturating_add(n_rows) > total_rows {
1892 return Err(DataError::SchemaMismatch {
1893 reason: "parquet row count changed while reading record batches".to_string(),
1894 });
1895 }
1896
1897 let decoded_columns = values
1898 .slice_mut(s![rows_seen..rows_seen + n_rows, ..])
1899 .axis_iter_mut(Axis(1))
1900 .into_par_iter()
1901 .zip(categorical_encoders.par_iter_mut())
1902 .zip(all_binary.par_iter_mut())
1903 .enumerate()
1904 .map(|(j, ((output, encoder), column_all_binary))| {
1905 decode_arrow_batch_column_into(
1906 batch.column(j).as_ref(),
1907 rows_seen,
1908 &headers[j],
1909 is_string_col[j],
1910 output,
1911 encoder.as_mut(),
1912 column_all_binary,
1913 )
1914 })
1915 .collect::<Vec<_>>();
1916
1917 for decoded in decoded_columns {
1922 decoded?;
1923 }
1924 rows_seen += n_rows;
1925 }
1926
1927 if rows_seen != total_rows {
1928 return Err(DataError::SchemaMismatch {
1929 reason: format!(
1930 "parquet metadata reports {total_rows} rows but record batches yielded {rows_seen}"
1931 ),
1932 });
1933 }
1934 let batches_ms = t_batches.elapsed().as_secs_f64() * 1000.0;
1935 if batches_ms > 100.0 {
1936 log::info!(
1937 "[DATA-LOAD] parquet_batches_decode | n_rows={} | n_cols={} | {:.1}ms",
1938 total_rows,
1939 p,
1940 batches_ms
1941 );
1942 }
1943 let t_schema = std::time::Instant::now();
1944 let mut levels = vec![Vec::<String>::new(); p];
1950 for (j, encoder) in categorical_encoders.into_iter().enumerate() {
1951 if let Some(encoder) = encoder {
1952 let order = if forced_numeric_categorical[j] {
1953 LevelOrder::Canonical
1954 } else {
1955 LevelOrder::Encounter
1956 };
1957 levels[j] = encoder.finish(values.column_mut(j), order);
1958 }
1959 }
1960 let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
1961 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
1962 for j in 0..p {
1963 let kind = if is_string_col[j] || forced_numeric_categorical[j] {
1964 ColumnKindTag::Categorical
1965 } else if all_binary[j] {
1966 ColumnKindTag::Binary
1967 } else {
1968 ColumnKindTag::Continuous
1969 };
1970 column_kinds.push(kind);
1971 schema_cols.push(SchemaColumn {
1972 name: headers[j].clone(),
1973 kind,
1974 levels: std::mem::take(&mut levels[j]),
1975 });
1976 }
1977 let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
1978 if schema_ms > 100.0 {
1979 let n_cat = column_kinds
1980 .iter()
1981 .filter(|k| matches!(k, ColumnKindTag::Categorical))
1982 .count();
1983 log::info!(
1984 "[DATA-LOAD] parquet_finalize_schema | n_cols={} | n_cat={} | {:.1}ms",
1985 p,
1986 n_cat,
1987 schema_ms
1988 );
1989 }
1990
1991 Ok(EncodedDataset {
1992 headers,
1993 values,
1994 schema: DataSchema {
1995 columns: schema_cols,
1996 },
1997 column_kinds,
1998 })
1999}
2000
2001fn load_parquet_with_schema(
2002 path: &Path,
2003 schema: &DataSchema,
2004 unseen_policy: UnseenCategoryPolicy,
2005 requested_columns: &[String],
2006) -> Result<EncodedDataset, DataError> {
2007 let inferred = load_parquet_inferred(path, requested_columns, &HashSet::new())?;
2011 let p = inferred.headers.len();
2012 let n = inferred.values.nrows();
2013
2014 let schema_byname: HashMap<&str, &SchemaColumn> = schema
2015 .columns
2016 .iter()
2017 .map(|c| (c.name.as_str(), c))
2018 .collect();
2019
2020 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2021 let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
2022 let mut values = inferred.values;
2023
2024 for j in 0..p {
2025 let name = &inferred.headers[j];
2026 if let Some(sc) = schema_byname.get(name.as_str()) {
2027 column_kinds.push(sc.kind);
2028 schema_cols.push((*sc).clone());
2029
2030 match sc.kind {
2031 ColumnKindTag::Continuous => {
2032 if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2033 return Err(DataError::SchemaMismatch {
2034 reason: format!(
2035 "column '{}' is continuous in schema but parquet column is string/categorical",
2036 name
2037 ),
2038 });
2039 }
2040 }
2041 ColumnKindTag::Binary => {
2042 if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2043 return Err(DataError::SchemaMismatch {
2044 reason: format!(
2045 "column '{}' is binary in schema but parquet column is string/categorical",
2046 name
2047 ),
2048 });
2049 }
2050 if let Some(row) = values.column(j).iter().position(|value| {
2051 (*value - 0.0).abs() >= 1e-12 && (*value - 1.0).abs() >= 1e-12
2052 }) {
2053 return Err(DataError::SchemaMismatch {
2054 reason: format!(
2055 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2056 name,
2057 row + 1,
2058 values[[row, j]]
2059 ),
2060 });
2061 }
2062 }
2063 ColumnKindTag::Categorical => {
2064 if !matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
2065 return Err(DataError::SchemaMismatch {
2066 reason: format!(
2067 "column '{}' is categorical in schema but parquet column is numeric",
2068 name
2069 ),
2070 });
2071 }
2072 let inferred_col = &inferred.schema.columns[j];
2073 let schema_level_map: HashMap<&str, f64> = sc
2075 .levels
2076 .iter()
2077 .enumerate()
2078 .map(|(idx, v)| (v.as_str(), idx as f64))
2079 .collect();
2080 let inferred_to_schema: Vec<f64> = inferred_col
2081 .levels
2082 .iter()
2083 .map(|lv| {
2084 schema_level_map
2085 .get(lv.as_str())
2086 .copied()
2087 .or_else(|| unseen_policy.unseen_code_for(name, sc.levels.len()))
2088 .ok_or_else(|| DataError::SchemaMismatch {
2089 reason: format!(
2090 "unseen level '{}' in categorical column '{}'",
2091 lv, name
2092 ),
2093 })
2094 })
2095 .collect::<Result<Vec<_>, _>>()?;
2096 for i in 0..n {
2097 let old_code = values[[i, j]] as usize;
2098 if old_code >= inferred_to_schema.len() {
2099 let Some(unseen_code) =
2100 unseen_policy.unseen_code_for(name, sc.levels.len())
2101 else {
2102 return Err(DataError::SchemaMismatch {
2103 reason: format!(
2104 "unseen categorical code at row {}, column '{}'",
2105 i + 1,
2106 name
2107 ),
2108 });
2109 };
2110 values[[i, j]] = unseen_code;
2111 continue;
2112 }
2113 values[[i, j]] = inferred_to_schema[old_code];
2114 }
2115 }
2116 }
2117 } else {
2118 column_kinds.push(inferred.column_kinds[j]);
2120 schema_cols.push(inferred.schema.columns[j].clone());
2121 }
2122 }
2123
2124 Ok(EncodedDataset {
2125 headers: inferred.headers,
2126 values,
2127 schema: DataSchema {
2128 columns: schema_cols,
2129 },
2130 column_kinds,
2131 })
2132}
2133
2134pub fn encode_recordswith_inferred_schema(
2135 headers: Vec<String>,
2136 records: Vec<StringRecord>,
2137) -> Result<EncodedDataset, String> {
2138 if records.is_empty() {
2139 return Err(DataError::EmptyInput {
2140 reason: "table data cannot be empty".to_string(),
2141 }
2142 .into());
2143 }
2144 let schema_cols = headers
2150 .par_iter()
2151 .enumerate()
2152 .map(|(j, name)| infer_schema_column(name, &records, j).map_err(String::from))
2153 .collect::<Result<Vec<SchemaColumn>, String>>()?;
2154 let schema = DataSchema {
2155 columns: schema_cols,
2156 };
2157 encode_recordswith_schema(headers, records, &schema, UnseenCategoryPolicy::Error)
2158}
2159
2160pub fn encode_recordswith_schema(
2161 headers: Vec<String>,
2162 records: Vec<StringRecord>,
2163 schema: &DataSchema,
2164 unseen_policy: UnseenCategoryPolicy,
2165) -> Result<EncodedDataset, String> {
2166 let n = records.len();
2167 if n == 0 {
2168 return Err(DataError::EmptyInput {
2169 reason: "table data cannot be empty".to_string(),
2170 }
2171 .into());
2172 }
2173 let p = headers.len();
2174 if p == 0 {
2175 return Err(DataError::EmptyInput {
2176 reason: "table data must have at least one header column".to_string(),
2177 }
2178 .into());
2179 }
2180 for (i, rec) in records.iter().enumerate() {
2187 if rec.len() != p {
2188 return Err(DataError::SchemaMismatch {
2189 reason: format!(
2190 "row width mismatch at row {}: got {} fields, expected {} (one per header)",
2191 i + 1,
2192 rec.len(),
2193 p
2194 ),
2195 }
2196 .into());
2197 }
2198 }
2199 let schema_byname: HashMap<&str, &SchemaColumn> = schema
2200 .columns
2201 .iter()
2202 .map(|c| (c.name.as_str(), c))
2203 .collect();
2204
2205 let encoded_columns = headers
2211 .par_iter()
2212 .enumerate()
2213 .map(|(j, name)| {
2214 let inferred_for_extra;
2215 let col_schema = if let Some(s) = schema_byname.get(name.as_str()) {
2216 *s
2217 } else {
2218 inferred_for_extra =
2219 infer_schema_column(name, &records, j).map_err(String::from)?;
2220 &inferred_for_extra
2221 };
2222 let column = encode_one_column(name, &records, j, col_schema, &unseen_policy)?;
2223 Ok::<(ColumnKindTag, Vec<f64>), String>((col_schema.kind, column))
2224 })
2225 .collect::<Result<Vec<(ColumnKindTag, Vec<f64>)>, String>>()?;
2226
2227 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
2228 let mut values = Array2::<f64>::zeros((n, p));
2229 for (j, (kind, column)) in encoded_columns.into_iter().enumerate() {
2230 column_kinds.push(kind);
2231 values
2232 .column_mut(j)
2233 .assign(&ndarray::ArrayView1::from(&column));
2234 }
2235
2236 Ok(EncodedDataset {
2237 headers,
2238 values,
2239 schema: schema.clone(),
2240 column_kinds,
2241 })
2242}
2243
2244fn encode_one_column(
2251 name: &str,
2252 records: &[StringRecord],
2253 j: usize,
2254 col_schema: &SchemaColumn,
2255 unseen_policy: &UnseenCategoryPolicy,
2256) -> Result<Vec<f64>, String> {
2257 let level_map = if matches!(col_schema.kind, ColumnKindTag::Categorical) {
2258 Some(
2259 col_schema
2260 .levels
2261 .iter()
2262 .enumerate()
2263 .map(|(idx, v)| (v.as_str(), idx as f64))
2264 .collect::<HashMap<_, _>>(),
2265 )
2266 } else {
2267 None
2268 };
2269
2270 let mut column = Vec::<f64>::with_capacity(records.len());
2271 for (i, rec) in records.iter().enumerate() {
2272 let raw = rec
2273 .get(j)
2274 .ok_or_else(|| {
2275 String::from(DataError::SchemaMismatch {
2276 reason: format!("missing field at row {}, col {}", i + 1, j + 1),
2277 })
2278 })?
2279 .trim();
2280 if raw.is_empty() {
2281 return Err(DataError::EmptyInput {
2282 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2283 }
2284 .into());
2285 }
2286 let val = match col_schema.kind {
2287 ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
2288 String::from(DataError::SchemaMismatch {
2289 reason: format!(
2290 "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
2291 name,
2292 i + 1,
2293 raw,
2294 err
2295 ),
2296 })
2297 })?,
2298 ColumnKindTag::Binary => {
2299 let v = raw.parse::<f64>().map_err(|err| {
2300 String::from(DataError::SchemaMismatch {
2301 reason: format!(
2302 "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
2303 name,
2304 i + 1,
2305 raw,
2306 err
2307 ),
2308 })
2309 })?;
2310 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2311 return Err(DataError::SchemaMismatch {
2312 reason: format!(
2313 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2314 name,
2315 i + 1,
2316 v
2317 ),
2318 }
2319 .into());
2320 }
2321 v
2322 }
2323 ColumnKindTag::Categorical => {
2324 let map = level_map.as_ref().ok_or_else(|| {
2325 String::from(DataError::EncodingFailure {
2326 reason: "internal categorical schema map missing".to_string(),
2327 })
2328 })?;
2329 match map.get(raw) {
2330 Some(v) => *v,
2331 None => unseen_policy
2332 .unseen_code_for(name, col_schema.levels.len())
2333 .ok_or_else(|| {
2334 String::from(DataError::SchemaMismatch {
2335 reason: format!(
2336 "unseen level '{}' in categorical column '{}' at row {}; allowed levels: {}",
2337 raw,
2338 name,
2339 i + 1,
2340 col_schema.levels.join(",")
2341 ),
2342 })
2343 })?,
2344 }
2345 }
2346 };
2347 if !val.is_finite() {
2348 return Err(DataError::InvalidValue {
2349 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2350 }
2351 .into());
2352 }
2353 column.push(val);
2354 }
2355 Ok(column)
2356}
2357
2358fn infer_schema_column(
2359 name: &str,
2360 records: &[StringRecord],
2361 col_idx: usize,
2362) -> Result<SchemaColumn, DataError> {
2363 let mut all_numeric = true;
2364 let mut all_binary = true;
2365 let mut levels = Vec::<String>::new();
2366 let mut level_index = HashMap::<String, usize>::new();
2367 for (i, rec) in records.iter().enumerate() {
2368 let raw = rec
2369 .get(col_idx)
2370 .ok_or_else(|| DataError::SchemaMismatch {
2371 reason: format!("missing field at row {}, col {}", i + 1, col_idx + 1),
2372 })?
2373 .trim();
2374 if raw.is_empty() {
2375 return Err(DataError::EmptyInput {
2376 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2377 });
2378 }
2379 if let Ok(v) = raw.parse::<f64>() {
2380 if !v.is_finite() {
2381 return Err(DataError::InvalidValue {
2382 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2383 });
2384 }
2385 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2386 all_binary = false;
2387 }
2388 } else {
2389 all_numeric = false;
2390 all_binary = false;
2391 level_index.entry(raw.to_string()).or_insert_with(|| {
2392 let idx = levels.len();
2393 levels.push(raw.to_string());
2394 idx
2395 });
2396 }
2397 }
2398 let kind = if all_numeric {
2399 if all_binary {
2400 ColumnKindTag::Binary
2401 } else {
2402 ColumnKindTag::Continuous
2403 }
2404 } else {
2405 ColumnKindTag::Categorical
2406 };
2407 if matches!(kind, ColumnKindTag::Categorical) {
2412 sort_levels_canonical(&mut levels);
2413 }
2414 Ok(SchemaColumn {
2415 name: name.to_string(),
2416 kind,
2417 levels: if matches!(kind, ColumnKindTag::Categorical) {
2418 levels
2419 } else {
2420 Vec::new()
2421 },
2422 })
2423}
2424
2425pub fn infer_and_encode_column_major(
2438 name: &str,
2439 column: &[&str],
2440 col_index: usize,
2441) -> Result<(SchemaColumn, Vec<f64>), String> {
2442 if column.is_empty() {
2443 return Err(DataError::EmptyInput {
2444 reason: "table data cannot be empty".to_string(),
2445 }
2446 .into());
2447 }
2448 let force_categorical = column.iter().any(|c| strip_categorical_sentinel(c).1);
2453 let mut all_numeric = !force_categorical;
2454 let mut all_binary = !force_categorical;
2455 let mut levels = Vec::<String>::new();
2456 let mut level_index = HashMap::<String, usize>::new();
2457 let mut trimmed = Vec::<&str>::with_capacity(column.len());
2458 let mut parsed = Vec::<Option<f64>>::with_capacity(column.len());
2465 for (i, raw_field) in column.iter().enumerate() {
2466 let (raw, _) = strip_categorical_sentinel(raw_field);
2469 let raw = raw.trim();
2470 if raw.is_empty() {
2471 return Err(DataError::EmptyInput {
2472 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2473 }
2474 .into());
2475 }
2476 if !force_categorical {
2479 if let Ok(v) = raw.parse::<f64>() {
2480 if !v.is_finite() {
2481 return Err(DataError::InvalidValue {
2482 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2483 }
2484 .into());
2485 }
2486 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2487 all_binary = false;
2488 }
2489 parsed.push(Some(v));
2490 trimmed.push(raw);
2491 continue;
2492 }
2493 all_numeric = false;
2494 all_binary = false;
2495 }
2496 level_index.entry(raw.to_string()).or_insert_with(|| {
2497 let idx = levels.len();
2498 levels.push(raw.to_string());
2499 idx
2500 });
2501 parsed.push(None);
2502 trimmed.push(raw);
2503 }
2504 let kind = if all_numeric {
2505 if all_binary {
2506 ColumnKindTag::Binary
2507 } else {
2508 ColumnKindTag::Continuous
2509 }
2510 } else {
2511 ColumnKindTag::Categorical
2512 };
2513 if matches!(kind, ColumnKindTag::Categorical) {
2527 sort_levels_canonical(&mut levels);
2528 }
2529 let schema = SchemaColumn {
2530 name: name.to_string(),
2531 kind,
2532 levels: if matches!(kind, ColumnKindTag::Categorical) {
2533 levels
2534 } else {
2535 Vec::new()
2536 },
2537 };
2538
2539 let level_map = if matches!(kind, ColumnKindTag::Categorical) {
2540 Some(
2541 schema
2542 .levels
2543 .iter()
2544 .enumerate()
2545 .map(|(idx, v)| (v.as_str(), idx as f64))
2546 .collect::<HashMap<_, _>>(),
2547 )
2548 } else {
2549 None
2550 };
2551
2552 let mut values = Vec::<f64>::with_capacity(trimmed.len());
2553 for (i, raw) in trimmed.iter().enumerate() {
2554 let raw = *raw;
2555 let val = match kind {
2556 ColumnKindTag::Continuous => parsed[i].ok_or_else(|| {
2560 String::from(DataError::EncodingFailure {
2561 reason: format!(
2562 "internal: continuous column '{}' lost its parsed value at row {} (col {})",
2563 name,
2564 i + 1,
2565 col_index
2566 ),
2567 })
2568 })?,
2569 ColumnKindTag::Binary => {
2570 let v = parsed[i].ok_or_else(|| {
2571 String::from(DataError::EncodingFailure {
2572 reason: format!(
2573 "internal: binary column '{}' lost its parsed value at row {} (col {})",
2574 name,
2575 i + 1,
2576 col_index
2577 ),
2578 })
2579 })?;
2580 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2581 return Err(DataError::SchemaMismatch {
2582 reason: format!(
2583 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2584 name,
2585 i + 1,
2586 v
2587 ),
2588 }
2589 .into());
2590 }
2591 v
2592 }
2593 ColumnKindTag::Categorical => {
2594 let map = level_map.as_ref().ok_or_else(|| {
2595 String::from(DataError::EncodingFailure {
2596 reason: "internal categorical schema map missing".to_string(),
2597 })
2598 })?;
2599 *map.get(raw).ok_or_else(|| {
2600 String::from(DataError::EncodingFailure {
2601 reason: format!(
2602 "internal: level '{}' missing from freshly built map for column '{}' (col {})",
2603 raw, name, col_index
2604 ),
2605 })
2606 })?
2607 }
2608 };
2609 if !val.is_finite() {
2610 return Err(DataError::InvalidValue {
2611 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2612 }
2613 .into());
2614 }
2615 values.push(val);
2616 }
2617 Ok((schema, values))
2618}
2619
2620#[cfg(test)]
2621mod tests {
2622 use super::*;
2623 use arrow::array::ArrayRef;
2624 use arrow::datatypes::{Field, Schema};
2625 use arrow::error::ArrowError;
2626 use arrow::record_batch::{RecordBatch, RecordBatchIterator};
2627 use std::sync::Arc;
2628
2629 fn encode_single_arrow_array(array: ArrayRef) -> Result<EncodedDataset, DataError> {
2630 let schema = Arc::new(Schema::new(vec![Field::new(
2631 "source",
2632 array.data_type().clone(),
2633 true,
2634 )]));
2635 let batch = RecordBatch::try_new(schema.clone(), vec![array]).expect("record batch");
2636 let batches: Vec<Result<RecordBatch, ArrowError>> = vec![Ok(batch)];
2637 let mut reader = RecordBatchIterator::new(batches, schema);
2638 encode_arrow_record_batch_reader_with_inferred_schema(
2639 &mut reader,
2640 vec!["normalized".to_string()],
2641 )
2642 }
2643
2644 #[test]
2645 fn arrow_reader_streams_typed_columns_in_supplied_order() {
2646 use arrow::array::{
2647 Array, BooleanArray, DictionaryArray, Float32Array, Int8Array, Int32Array, Int64Array,
2648 LargeStringArray, StringArray,
2649 };
2650 use arrow::datatypes::Int8Type;
2651
2652 let string_dictionary_1: DictionaryArray<Int8Type> =
2653 vec!["beta", "alpha"].into_iter().collect();
2654 let string_dictionary_2: DictionaryArray<Int8Type> = vec!["beta"].into_iter().collect();
2655 let numeric_dictionary_1 = DictionaryArray::<Int8Type>::new(
2656 Int8Array::from(vec![0, 1]),
2657 Arc::new(Int64Array::from(vec![5, 7])),
2658 );
2659 let numeric_dictionary_2 = DictionaryArray::<Int8Type>::new(
2660 Int8Array::from(vec![0]),
2661 Arc::new(Int64Array::from(vec![5])),
2662 );
2663
2664 let schema = Arc::new(Schema::new(vec![
2665 Field::new("source_float", arrow::datatypes::DataType::Float32, false),
2666 Field::new("source_integer", arrow::datatypes::DataType::Int32, false),
2667 Field::new("source_flag", arrow::datatypes::DataType::Boolean, false),
2668 Field::new("source_utf8", arrow::datatypes::DataType::Utf8, false),
2669 Field::new(
2670 "source_large_utf8",
2671 arrow::datatypes::DataType::LargeUtf8,
2672 false,
2673 ),
2674 Field::new(
2675 "source_dictionary_string",
2676 string_dictionary_1.data_type().clone(),
2677 false,
2678 ),
2679 Field::new(
2680 "source_dictionary_number",
2681 numeric_dictionary_1.data_type().clone(),
2682 false,
2683 ),
2684 ]));
2685 let batch_1 = RecordBatch::try_new(
2686 schema.clone(),
2687 vec![
2688 Arc::new(Float32Array::from(vec![1.5, 2.5])) as ArrayRef,
2689 Arc::new(Int32Array::from(vec![0, 1])),
2690 Arc::new(BooleanArray::from(vec![true, false])),
2691 Arc::new(StringArray::from(vec!["item10", "item2"])),
2692 Arc::new(LargeStringArray::from(vec!["z", "a"])),
2693 Arc::new(string_dictionary_1),
2694 Arc::new(numeric_dictionary_1),
2695 ],
2696 )
2697 .expect("first record batch");
2698 let batch_2 = RecordBatch::try_new(
2699 schema.clone(),
2700 vec![
2701 Arc::new(Float32Array::from(vec![-4.0])) as ArrayRef,
2702 Arc::new(Int32Array::from(vec![1])),
2703 Arc::new(BooleanArray::from(vec![true])),
2704 Arc::new(StringArray::from(vec!["item1"])),
2705 Arc::new(LargeStringArray::from(vec!["z"])),
2706 Arc::new(string_dictionary_2),
2707 Arc::new(numeric_dictionary_2),
2708 ],
2709 )
2710 .expect("second record batch");
2711 let batches: Vec<Result<RecordBatch, ArrowError>> = vec![Ok(batch_1), Ok(batch_2)];
2712 let mut reader = RecordBatchIterator::new(batches, schema);
2713 let headers = [
2714 "float",
2715 "integer",
2716 "flag",
2717 "utf8",
2718 "large_utf8",
2719 "dictionary_string",
2720 "dictionary_number",
2721 ]
2722 .map(str::to_string)
2723 .to_vec();
2724
2725 let dataset =
2726 encode_arrow_record_batch_reader_with_inferred_schema(&mut reader, headers.clone())
2727 .expect("Arrow stream should encode");
2728
2729 assert_eq!(dataset.headers, headers);
2730 assert_eq!(
2731 dataset.column_kinds,
2732 vec![
2733 ColumnKindTag::Continuous,
2734 ColumnKindTag::Binary,
2735 ColumnKindTag::Binary,
2736 ColumnKindTag::Categorical,
2737 ColumnKindTag::Categorical,
2738 ColumnKindTag::Categorical,
2739 ColumnKindTag::Continuous,
2740 ]
2741 );
2742 assert_eq!(
2743 dataset.values,
2744 ndarray::arr2(&[
2745 [1.5, 0.0, 1.0, 2.0, 1.0, 1.0, 5.0],
2746 [2.5, 1.0, 0.0, 1.0, 0.0, 0.0, 7.0],
2747 [-4.0, 1.0, 1.0, 0.0, 1.0, 1.0, 5.0],
2748 ])
2749 );
2750 assert_eq!(
2751 dataset.schema.columns[3].levels,
2752 vec!["item1", "item2", "item10"]
2753 );
2754 assert_eq!(dataset.schema.columns[4].levels, vec!["a", "z"]);
2755 assert_eq!(dataset.schema.columns[5].levels, vec!["alpha", "beta"]);
2756 assert!(
2757 dataset
2758 .schema
2759 .columns
2760 .iter()
2761 .zip(dataset.headers.iter())
2762 .all(|(column, header)| column.name == *header)
2763 );
2764 }
2765
2766 #[test]
2767 fn arrow_reader_rejects_empty_duplicate_and_mismatched_headers() {
2768 let schema = Arc::new(Schema::new(vec![Field::new(
2769 "source",
2770 arrow::datatypes::DataType::Int32,
2771 false,
2772 )]));
2773
2774 let mut empty_name_reader = RecordBatchIterator::new(
2775 Vec::<Result<RecordBatch, ArrowError>>::new(),
2776 schema.clone(),
2777 );
2778 let empty_name = encode_arrow_record_batch_reader_with_inferred_schema(
2779 &mut empty_name_reader,
2780 vec![" ".to_string()],
2781 )
2782 .expect_err("blank header should fail");
2783 assert!(matches!(empty_name, DataError::EmptyInput { .. }));
2784
2785 let mut duplicate_reader = RecordBatchIterator::new(
2786 Vec::<Result<RecordBatch, ArrowError>>::new(),
2787 Arc::new(Schema::new(vec![
2788 Field::new("a", arrow::datatypes::DataType::Int32, false),
2789 Field::new("b", arrow::datatypes::DataType::Int32, false),
2790 ])),
2791 );
2792 let duplicate = encode_arrow_record_batch_reader_with_inferred_schema(
2793 &mut duplicate_reader,
2794 vec!["x".to_string(), "x".to_string()],
2795 )
2796 .expect_err("duplicate header should fail");
2797 assert!(matches!(duplicate, DataError::SchemaMismatch { .. }));
2798
2799 let mut mismatch_reader =
2800 RecordBatchIterator::new(Vec::<Result<RecordBatch, ArrowError>>::new(), schema);
2801 let mismatch = encode_arrow_record_batch_reader_with_inferred_schema(
2802 &mut mismatch_reader,
2803 vec!["x".to_string(), "y".to_string()],
2804 )
2805 .expect_err("header count mismatch should fail");
2806 assert!(matches!(mismatch, DataError::SchemaMismatch { .. }));
2807 }
2808
2809 #[test]
2810 fn arrow_reader_reports_typed_null_nonfinite_and_unsupported_errors() {
2811 use arrow::array::{Date32Array, DictionaryArray, Float64Array, Int8Array, StringArray};
2812 use arrow::datatypes::Int8Type;
2813
2814 let null_numeric =
2815 encode_single_arrow_array(Arc::new(Float64Array::from(vec![Some(1.0), None])))
2816 .expect_err("numeric null should fail");
2817 assert!(matches!(&null_numeric, DataError::InvalidValue { .. }));
2818 assert!(null_numeric.to_string().contains("null value at row 2"));
2819
2820 let null_dictionary = DictionaryArray::<Int8Type>::new(
2821 Int8Array::from(vec![0, 1]),
2822 Arc::new(StringArray::from(vec![Some("present"), None])),
2823 );
2824 let logical_null = encode_single_arrow_array(Arc::new(null_dictionary))
2825 .expect_err("null dictionary value should fail");
2826 assert!(matches!(&logical_null, DataError::InvalidValue { .. }));
2827 assert!(logical_null.to_string().contains("null value at row 2"));
2828
2829 let nonfinite = encode_single_arrow_array(Arc::new(Float64Array::from(vec![f64::NAN])))
2830 .expect_err("NaN should fail");
2831 assert!(matches!(&nonfinite, DataError::InvalidValue { .. }));
2832 assert!(nonfinite.to_string().contains("non-finite value"));
2833
2834 let unsupported = encode_single_arrow_array(Arc::new(Date32Array::from(vec![1])))
2835 .expect_err("date column should fail");
2836 assert!(matches!(&unsupported, DataError::InvalidValue { .. }));
2837 assert!(
2838 unsupported
2839 .to_string()
2840 .contains("unsupported Arrow column type")
2841 );
2842 }
2843
2844 #[test]
2845 fn encode_records_rejects_empty_input() {
2846 let headers = vec!["x".to_string()];
2847 let schema = DataSchema {
2848 columns: vec![SchemaColumn {
2849 name: "x".to_string(),
2850 kind: ColumnKindTag::Continuous,
2851 levels: Vec::new(),
2852 }],
2853 };
2854
2855 let err = encode_recordswith_inferred_schema(headers.clone(), Vec::new())
2856 .expect_err("empty inferred records should error");
2857 assert_eq!(err, "table data cannot be empty");
2858
2859 let err =
2860 encode_recordswith_schema(headers, Vec::new(), &schema, UnseenCategoryPolicy::Error)
2861 .expect_err("empty schema-guided records should error");
2862 assert_eq!(err, "table data cannot be empty");
2863 }
2864
2865 #[test]
2866 fn column_major_matches_record_driven_inferred_encode() {
2867 let headers = vec!["cont".to_string(), "bin".to_string(), "cat".to_string()];
2872 let raw_rows = vec![
2873 vec!["1.5", "0", "a"],
2874 vec!["2.0", "1", "b"],
2875 vec!["-3.25", "1", "a"],
2876 vec!["0.0", "0", "c"],
2877 ];
2878 let records: Vec<StringRecord> = raw_rows
2879 .iter()
2880 .map(|r| StringRecord::from(r.clone()))
2881 .collect();
2882 let record_ds = encode_recordswith_inferred_schema(headers.clone(), records)
2883 .expect("record-driven encode");
2884
2885 for (j, name) in headers.iter().enumerate() {
2886 let column: Vec<&str> = raw_rows.iter().map(|r| r[j]).collect();
2887 let (schema_col, values) =
2888 infer_and_encode_column_major(name, &column, j + 1).expect("column-major encode");
2889 assert_eq!(schema_col.kind, record_ds.schema.columns[j].kind);
2890 assert_eq!(schema_col.levels, record_ds.schema.columns[j].levels);
2891 for (i, v) in values.iter().enumerate() {
2892 assert_eq!(*v, record_ds.values[[i, j]], "row {i} col {name}");
2893 }
2894 }
2895 }
2896
2897 #[test]
2898 fn encode_records_can_encode_unseen_named_categorical_column() {
2899 let schema = DataSchema {
2900 columns: vec![
2901 SchemaColumn {
2902 name: "g".to_string(),
2903 kind: ColumnKindTag::Categorical,
2904 levels: vec!["a".to_string(), "b".to_string()],
2905 },
2906 SchemaColumn {
2907 name: "x".to_string(),
2908 kind: ColumnKindTag::Categorical,
2909 levels: vec!["low".to_string(), "high".to_string()],
2910 },
2911 ],
2912 };
2913 let headers = vec!["g".to_string(), "x".to_string()];
2914 let records = vec![StringRecord::from(vec!["new-group", "low"])];
2915 let policy =
2916 UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
2917
2918 let ds =
2919 encode_recordswith_schema(headers, records, &schema, policy).expect("encoded dataset");
2920
2921 assert_eq!(ds.values[[0, 0]], 2.0);
2922 assert_eq!(ds.values[[0, 1]], 0.0);
2923 }
2924
2925 #[test]
2926 fn categorical_encoder_consumes_labels_and_remaps_canonically() {
2927 use ndarray::Array1;
2928
2929 let mut encoder = CategoricalEncoder::default();
2930 let mut encoded = Array1::from_vec(
2931 ["item10", "item2", "item1", "item2"]
2932 .into_iter()
2933 .map(|label| encoder.encode(label) as f64)
2934 .collect(),
2935 );
2936
2937 let levels = encoder.finish(encoded.view_mut(), LevelOrder::Canonical);
2938
2939 assert_eq!(levels, vec!["item1", "item2", "item10"]);
2940 assert_eq!(encoded.to_vec(), vec![2.0, 1.0, 0.0, 1.0]);
2941 }
2942
2943 #[test]
2944 fn complete_delimited_schema_encodes_projected_rows_directly() {
2945 let dir = tempfile::tempdir().expect("tempdir");
2946 let path = dir.path().join("schema_direct.csv");
2947 std::fs::write(
2948 &path,
2949 "y,group,flag,unused\n1.5,b,0,first\n2.5,a,1,second\n",
2950 )
2951 .expect("write csv");
2952 let schema = DataSchema {
2953 columns: vec![
2954 SchemaColumn {
2955 name: "group".to_string(),
2956 kind: ColumnKindTag::Categorical,
2957 levels: vec!["a".to_string(), "b".to_string()],
2958 },
2959 SchemaColumn {
2960 name: "flag".to_string(),
2961 kind: ColumnKindTag::Binary,
2962 levels: Vec::new(),
2963 },
2964 SchemaColumn {
2965 name: "y".to_string(),
2966 kind: ColumnKindTag::Continuous,
2967 levels: Vec::new(),
2968 },
2969 ],
2970 };
2971
2972 let loaded = load_datasetwith_schema_projected(
2973 &path,
2974 &schema,
2975 UnseenCategoryPolicy::Error,
2976 &["y".to_string(), "group".to_string(), "flag".to_string()],
2977 )
2978 .expect("schema-guided projected load");
2979
2980 assert_eq!(loaded.headers, vec!["y", "group", "flag"]);
2981 assert_eq!(loaded.values.row(0).to_vec(), vec![1.5, 1.0, 0.0]);
2982 assert_eq!(loaded.values.row(1).to_vec(), vec![2.5, 0.0, 1.0]);
2983 assert_eq!(
2984 loaded.column_kinds,
2985 vec![
2986 ColumnKindTag::Continuous,
2987 ColumnKindTag::Categorical,
2988 ColumnKindTag::Binary,
2989 ]
2990 );
2991 }
2992
2993 #[test]
2994 fn direct_parquet_decoder_preserves_string_encounter_order() {
2995 use arrow::array::DictionaryArray;
2996 use arrow::datatypes::Int8Type;
2997 use ndarray::Array1;
2998
2999 let dictionary: DictionaryArray<Int8Type> =
3000 vec!["beta", "alpha", "beta"].into_iter().collect();
3001 let mut encoded = Array1::<f64>::zeros(dictionary.len());
3002 let mut encoder = CategoricalEncoder::default();
3003 let mut all_binary = true;
3004
3005 decode_arrow_batch_column_into(
3006 &dictionary,
3007 0,
3008 "group",
3009 true,
3010 encoded.view_mut(),
3011 Some(&mut encoder),
3012 &mut all_binary,
3013 )
3014 .expect("dictionary strings decode directly");
3015 let levels = encoder.finish(encoded.view_mut(), LevelOrder::Encounter);
3016
3017 assert_eq!(levels, vec!["beta", "alpha"]);
3018 assert_eq!(encoded.to_vec(), vec![0.0, 1.0, 0.0]);
3019 }
3020
3021 #[test]
3022 fn numeric_valued_dictionary_column_classifies_and_decodes_as_numeric() {
3023 use arrow::array::{Array, ArrayRef, DictionaryArray, Int8Array, Int64Array};
3033 use arrow::datatypes::{DataType, Int8Type};
3034 use std::sync::Arc;
3035
3036 let keys = Int8Array::from(vec![0i8, 1, 0, 1, 0]);
3038 let dict_values: ArrayRef = Arc::new(Int64Array::from(vec![5i64, 7]));
3039 let dict: DictionaryArray<Int8Type> = DictionaryArray::new(keys, dict_values);
3040
3041 assert!(matches!(dict.data_type(), DataType::Dictionary(_, _)));
3044 assert!(
3045 !arrow_field_is_string(dict.data_type()),
3046 "Dictionary(Int8, Int64) must not be treated as a string column"
3047 );
3048
3049 let str_dict: DictionaryArray<Int8Type> = vec!["a", "b", "a"].into_iter().collect();
3051 assert!(
3052 arrow_field_is_string(str_dict.data_type()),
3053 "Dictionary(Int8, Utf8) must remain a string column"
3054 );
3055
3056 let mut decoded = ndarray::Array1::<f64>::zeros(dict.len());
3060 let mut all_binary = true;
3061 decode_arrow_batch_column_into(
3062 &dict,
3063 0,
3064 "x",
3065 false,
3066 decoded.view_mut(),
3067 None,
3068 &mut all_binary,
3069 )
3070 .expect("numeric dictionary column should decode as numeric");
3071 assert_eq!(decoded.to_vec(), vec![5.0, 7.0, 5.0, 7.0, 5.0]);
3072 assert!(!all_binary);
3073
3074 use arrow::datatypes::{Field, Schema};
3079 use arrow::record_batch::RecordBatch;
3080 use parquet::arrow::ArrowWriter;
3081
3082 let arrow_schema = Arc::new(Schema::new(vec![Field::new(
3083 "x",
3084 dict.data_type().clone(),
3085 false,
3086 )]));
3087 let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(dict.clone())])
3088 .expect("record batch with a dictionary numeric column");
3089
3090 let dir = tempfile::tempdir().expect("tempdir");
3091 let path = dir.path().join("dict_numeric.parquet");
3092 {
3093 let file = std::fs::File::create(&path).expect("create parquet");
3094 let mut writer =
3095 ArrowWriter::try_new(file, arrow_schema, None).expect("arrow parquet writer");
3096 writer.write(&batch).expect("write batch");
3097 writer.close().expect("close writer");
3098 }
3099
3100 let inferred =
3103 load_parquet_inferred(&path, &[], &HashSet::new()).expect("inferred parquet load");
3104 assert_eq!(inferred.column_kinds, vec![ColumnKindTag::Continuous]);
3105 assert_eq!(
3106 inferred.values.column(0).to_vec(),
3107 vec![5.0, 7.0, 5.0, 7.0, 5.0]
3108 );
3109
3110 let schema = DataSchema {
3114 columns: vec![SchemaColumn {
3115 name: "x".to_string(),
3116 kind: ColumnKindTag::Continuous,
3117 levels: Vec::new(),
3118 }],
3119 };
3120 let schema_loaded =
3121 load_parquet_with_schema(&path, &schema, UnseenCategoryPolicy::Error, &[])
3122 .expect("dictionary-encoded numeric parquet must load against a Continuous schema");
3123 assert_eq!(schema_loaded.column_kinds, vec![ColumnKindTag::Continuous]);
3124 assert_eq!(
3125 schema_loaded.values.column(0).to_vec(),
3126 vec![5.0, 7.0, 5.0, 7.0, 5.0]
3127 );
3128 }
3129
3130 #[test]
3131 fn encode_records_keeps_unlisted_categorical_columns_strict() {
3132 let schema = DataSchema {
3133 columns: vec![
3134 SchemaColumn {
3135 name: "g".to_string(),
3136 kind: ColumnKindTag::Categorical,
3137 levels: vec!["a".to_string(), "b".to_string()],
3138 },
3139 SchemaColumn {
3140 name: "x".to_string(),
3141 kind: ColumnKindTag::Categorical,
3142 levels: vec!["low".to_string(), "high".to_string()],
3143 },
3144 ],
3145 };
3146 let headers = vec!["g".to_string(), "x".to_string()];
3147 let records = vec![StringRecord::from(vec!["a", "new-level"])];
3148 let policy =
3149 UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
3150
3151 let err = encode_recordswith_schema(headers, records, &schema, policy)
3152 .expect_err("ordinary categorical column should stay strict");
3153
3154 assert!(err.contains("unseen level 'new-level' in categorical column 'x'"));
3155 }
3156
3157 #[test]
3162 fn sentinel_strip_present_returns_rest_and_true() {
3163 let marked = format!("{}{}", CATEGORICAL_CELL_SENTINEL, "hello");
3164 let (rest, found) = strip_categorical_sentinel(&marked);
3165 assert_eq!(rest, "hello");
3166 assert!(found);
3167 }
3168
3169 #[test]
3170 fn sentinel_strip_absent_returns_original_and_false() {
3171 let (rest, found) = strip_categorical_sentinel("hello");
3172 assert_eq!(rest, "hello");
3173 assert!(!found);
3174 }
3175
3176 #[test]
3177 fn sentinel_strip_empty_string_returns_empty_and_false() {
3178 let (rest, found) = strip_categorical_sentinel("");
3179 assert_eq!(rest, "");
3180 assert!(!found);
3181 }
3182
3183 #[test]
3184 fn sentinel_strip_only_sentinel_returns_empty_and_true() {
3185 let marked = CATEGORICAL_CELL_SENTINEL.to_string();
3186 let (rest, found) = strip_categorical_sentinel(&marked);
3187 assert_eq!(rest, "");
3188 assert!(found);
3189 }
3190
3191 #[test]
3196 fn feature_ranges_two_columns() {
3197 let values = ndarray::arr2(&[[1.0_f64, 10.0], [3.0, 20.0], [2.0, 15.0]]);
3198 let ds = EncodedDataset {
3199 headers: vec!["a".to_string(), "b".to_string()],
3200 values,
3201 schema: DataSchema { columns: vec![] },
3202 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3203 };
3204 let ranges = ds.feature_ranges();
3205 assert_eq!(ranges.len(), 2);
3206 assert_eq!(ranges[0], (1.0, 3.0));
3207 assert_eq!(ranges[1], (10.0, 20.0));
3208 }
3209
3210 #[test]
3211 fn feature_ranges_single_row_min_equals_max() {
3212 let values = ndarray::arr2(&[[5.0_f64, -3.0]]);
3213 let ds = EncodedDataset {
3214 headers: vec!["x".to_string(), "y".to_string()],
3215 values,
3216 schema: DataSchema { columns: vec![] },
3217 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3218 };
3219 let ranges = ds.feature_ranges();
3220 assert_eq!(ranges[0], (5.0, 5.0));
3221 assert_eq!(ranges[1], (-3.0, -3.0));
3222 }
3223
3224 #[test]
3225 fn feature_ranges_all_nan_defaults_to_zero() {
3226 let values = ndarray::arr2(&[[f64::NAN], [f64::NAN]]);
3227 let ds = EncodedDataset {
3228 headers: vec!["x".to_string()],
3229 values,
3230 schema: DataSchema { columns: vec![] },
3231 column_kinds: vec![ColumnKindTag::Continuous],
3232 };
3233 let ranges = ds.feature_ranges();
3234 assert_eq!(ranges[0], (0.0, 0.0));
3235 }
3236
3237 #[test]
3242 fn column_map_indexes_by_name() {
3243 let values = ndarray::arr2(&[[0.0_f64, 1.0], [2.0, 3.0]]);
3244 let ds = EncodedDataset {
3245 headers: vec!["alpha".to_string(), "beta".to_string()],
3246 values,
3247 schema: DataSchema { columns: vec![] },
3248 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
3249 };
3250 let map = ds.column_map();
3251 assert_eq!(map["alpha"], 0);
3252 assert_eq!(map["beta"], 1);
3253 assert_eq!(map.len(), 2);
3254 }
3255
3256 #[test]
3259 fn shared_prefix_identical_strings() {
3260 assert_eq!(shared_prefix("hello", "hello"), 5);
3261 }
3262
3263 #[test]
3264 fn shared_prefix_no_common_prefix() {
3265 assert_eq!(shared_prefix("abc", "xyz"), 0);
3266 }
3267
3268 #[test]
3269 fn shared_prefix_partial_match() {
3270 assert_eq!(shared_prefix("foobar", "foobaz"), 5);
3271 }
3272
3273 #[test]
3274 fn shared_prefix_one_empty() {
3275 assert_eq!(shared_prefix("", "hello"), 0);
3276 assert_eq!(shared_prefix("hello", ""), 0);
3277 }
3278
3279 #[test]
3280 fn shared_prefix_both_empty() {
3281 assert_eq!(shared_prefix("", ""), 0);
3282 }
3283
3284 #[test]
3285 fn shared_prefix_shorter_string_is_prefix() {
3286 assert_eq!(shared_prefix("foo", "foobar"), 3);
3287 }
3288
3289 #[test]
3292 fn detect_format_csv() {
3293 let path = std::path::Path::new("data.csv");
3294 assert_eq!(detect_format(path).unwrap(), DataFormat::Csv);
3295 }
3296
3297 #[test]
3298 fn detect_format_tsv() {
3299 assert_eq!(
3300 detect_format(std::path::Path::new("data.tsv")).unwrap(),
3301 DataFormat::Tsv
3302 );
3303 assert_eq!(
3304 detect_format(std::path::Path::new("data.txt")).unwrap(),
3305 DataFormat::Tsv
3306 );
3307 assert_eq!(
3308 detect_format(std::path::Path::new("data.tab")).unwrap(),
3309 DataFormat::Tsv
3310 );
3311 }
3312
3313 #[test]
3314 fn detect_format_parquet() {
3315 assert_eq!(
3316 detect_format(std::path::Path::new("data.parquet")).unwrap(),
3317 DataFormat::Parquet
3318 );
3319 assert_eq!(
3320 detect_format(std::path::Path::new("data.pq")).unwrap(),
3321 DataFormat::Parquet
3322 );
3323 assert_eq!(
3324 detect_format(std::path::Path::new("data.pqt")).unwrap(),
3325 DataFormat::Parquet
3326 );
3327 }
3328
3329 #[test]
3330 fn detect_format_uppercase_extension() {
3331 assert_eq!(
3332 detect_format(std::path::Path::new("data.CSV")).unwrap(),
3333 DataFormat::Csv
3334 );
3335 }
3336
3337 #[test]
3338 fn detect_format_unknown_extension_is_error() {
3339 let err = detect_format(std::path::Path::new("data.json")).unwrap_err();
3340 let msg = format!("{err:?}");
3341 assert!(
3342 msg.contains("json") || msg.contains("unsupported"),
3343 "error should mention extension, got: {msg}"
3344 );
3345 }
3346
3347 #[test]
3350 fn strip_categorical_sentinel_marked_cell() {
3351 let marked = "\u{0}hello";
3353 let (text, found) = strip_categorical_sentinel(marked);
3354 assert!(found);
3355 assert_eq!(text, "hello");
3356 }
3357
3358 #[test]
3359 fn strip_categorical_sentinel_unmarked_cell() {
3360 let (text, found) = strip_categorical_sentinel("plain");
3361 assert!(!found);
3362 assert_eq!(text, "plain");
3363 }
3364
3365 #[test]
3366 fn strip_categorical_sentinel_empty_string() {
3367 let (text, found) = strip_categorical_sentinel("");
3368 assert!(!found);
3369 assert_eq!(text, "");
3370 }
3371
3372 #[test]
3373 fn strip_categorical_sentinel_only_sentinel() {
3374 let s = "\u{0}";
3375 let (text, found) = strip_categorical_sentinel(s);
3376 assert!(found);
3377 assert_eq!(text, "");
3378 }
3379
3380 #[test]
3383 fn projected_headers_selects_by_index() {
3384 let all = vec![
3385 "a".to_string(),
3386 "b".to_string(),
3387 "c".to_string(),
3388 "d".to_string(),
3389 ];
3390 let selected = projected_headers(&all, &[1, 3]);
3391 assert_eq!(selected, vec!["b".to_string(), "d".to_string()]);
3392 }
3393
3394 #[test]
3395 fn projected_headers_empty_selection() {
3396 let all = vec!["x".to_string(), "y".to_string()];
3397 let selected = projected_headers(&all, &[]);
3398 assert!(selected.is_empty());
3399 }
3400
3401 #[test]
3402 fn projected_headers_all_indices() {
3403 let all = vec!["p".to_string(), "q".to_string()];
3404 let selected = projected_headers(&all, &[0, 1]);
3405 assert_eq!(selected, all);
3406 }
3407
3408 #[test]
3409 fn canonical_level_bits_collapses_signed_zero() {
3410 let pos = 0.0_f64;
3414 let neg = -0.0_f64;
3415 assert_ne!(
3416 pos.to_bits(),
3417 neg.to_bits(),
3418 "precondition: raw bits differ"
3419 );
3420 assert_eq!(pos, neg, "precondition: numerically equal");
3421 assert_eq!(canonical_level_bits(pos), canonical_level_bits(neg));
3422 assert_eq!(canonical_level_bits(neg), 0.0_f64.to_bits());
3423 assert_eq!(canonical_level_bits(-1.0 * 0.0), 0.0_f64.to_bits());
3425 assert_eq!(canonical_level_bits(0.0 - 0.0), 0.0_f64.to_bits());
3426 }
3427
3428 #[test]
3429 fn canonical_level_bits_is_bit_stable_on_ordinary_values() {
3430 for &v in &[
3433 1.0_f64,
3434 -1.0,
3435 2.5,
3436 -3.75,
3437 1e300,
3438 -1e-300,
3439 f64::MIN,
3440 f64::MAX,
3441 ] {
3442 assert_eq!(canonical_level_bits(v), v.to_bits(), "value {v}");
3443 }
3444 assert_ne!(canonical_level_bits(1.0), canonical_level_bits(2.0));
3446 assert_ne!(canonical_level_bits(0.0), canonical_level_bits(1.0));
3447 assert_ne!(
3449 canonical_level_bits(f64::INFINITY),
3450 canonical_level_bits(f64::NEG_INFINITY)
3451 );
3452 }
3453
3454 #[test]
3455 fn canonical_level_bits_collapses_nan_payloads() {
3456 let a = f64::NAN;
3458 let b = f64::from_bits(0x7ff8_0000_0000_0001); let c = -f64::NAN; assert!(a.is_nan() && b.is_nan() && c.is_nan());
3461 assert_eq!(canonical_level_bits(a), canonical_level_bits(b));
3462 assert_eq!(canonical_level_bits(a), canonical_level_bits(c));
3463 }
3464
3465 #[test]
3466 fn canonical_level_bits_is_idempotent() {
3467 for &v in &[0.0_f64, -0.0, 1.0, -2.0, f64::NAN] {
3470 let once = canonical_level_bits(v);
3471 let twice = canonical_level_bits(f64::from_bits(once));
3472 assert_eq!(once, twice, "value {v}");
3473 }
3474 }
3475}