1use csv::{ReaderBuilder, StringRecord};
2use ndarray::{Array2, Axis};
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 },
127 ColumnNotFound {
134 name: String,
136 role: Option<String>,
140 available: Vec<String>,
142 similar: Vec<String>,
145 tsv_hint: bool,
150 },
151}
152
153impl DataError {
154 pub fn column_not_found(
160 col_map: &HashMap<String, usize>,
161 name: &str,
162 role: Option<&str>,
163 ) -> Self {
164 let target_lower = name.to_lowercase();
165 let mut similar: Vec<String> = col_map
166 .keys()
167 .filter(|k| {
168 let k_lower = k.to_lowercase();
169 k_lower.contains(&target_lower)
170 || target_lower.contains(&k_lower)
171 || shared_prefix(&k_lower, &target_lower) >= 3
172 })
173 .cloned()
174 .collect();
175 similar.sort_unstable();
176 let mut available: Vec<String> = col_map.keys().cloned().collect();
177 available.sort_unstable();
178 let tsv_hint = available.len() == 1 && available[0].contains('\t');
179 Self::ColumnNotFound {
180 name: name.to_string(),
181 role: role.map(str::to_string),
182 available,
183 similar,
184 tsv_hint,
185 }
186 }
187}
188
189impl fmt::Display for DataError {
190 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
191 match self {
192 DataError::SchemaMismatch { reason }
193 | DataError::ParseError { reason }
194 | DataError::EncodingFailure { reason }
195 | DataError::EmptyInput { reason }
196 | DataError::InvalidValue { reason } => f.write_str(reason),
197 DataError::ColumnNotFound {
198 name,
199 role,
200 available,
201 similar,
202 tsv_hint,
203 } => {
204 let label = match role {
205 Some(r) => format!("{r} column '{name}'"),
206 None => format!("column '{name}'"),
207 };
208 let tsv_suffix = if *tsv_hint {
209 " — your file appears to be tab-separated; gam expects comma-separated CSV. \
210 Replace tabs with commas, or pre-convert with `tr '\\t' ',' < file.tsv > file.csv`."
211 } else {
212 ""
213 };
214 if similar.is_empty() {
215 write!(
216 f,
217 "{label} not found in data. Available columns: [{}]{tsv_suffix}",
218 available.join(", ")
219 )
220 } else {
221 write!(
222 f,
223 "{label} not found in data. Did you mean one of [{}]? Full list: [{}]{tsv_suffix}",
224 similar.join(", "),
225 available.join(", ")
226 )
227 }
228 }
229 }
230 }
231}
232
233impl std::error::Error for DataError {}
234
235impl From<DataError> for String {
236 fn from(err: DataError) -> String {
237 err.to_string()
238 }
239}
240
241#[derive(Clone, Debug, Serialize, Deserialize)]
246pub struct DataSchema {
247 pub columns: Vec<SchemaColumn>,
248}
249
250#[derive(Clone, Debug, Serialize, Deserialize)]
251pub struct SchemaColumn {
252 pub name: String,
253 pub kind: ColumnKindTag,
254 #[serde(default)]
255 pub levels: Vec<String>,
256}
257
258#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
259#[serde(rename_all = "kebab-case")]
260pub enum ColumnKindTag {
261 Continuous,
262 Binary,
263 Categorical,
264}
265
266#[derive(Clone, Debug, Eq, PartialEq)]
267pub enum UnseenCategoryPolicy {
268 Error,
269 EncodeUnknownForColumns(HashSet<String>),
270}
271
272impl UnseenCategoryPolicy {
273 pub fn encode_unknown_for_columns(columns: HashSet<String>) -> Self {
274 if columns.is_empty() {
275 Self::Error
276 } else {
277 Self::EncodeUnknownForColumns(columns)
278 }
279 }
280
281 fn unseen_code_for(&self, column_name: &str, level_count: usize) -> Option<f64> {
282 match self {
283 Self::Error => None,
284 Self::EncodeUnknownForColumns(columns) => {
285 columns.contains(column_name).then_some(level_count as f64)
286 }
287 }
288 }
289}
290
291#[derive(Clone, Debug)]
292pub struct EncodedDataset {
293 pub headers: Vec<String>,
294 pub values: Array2<f64>,
295 pub schema: DataSchema,
296 pub column_kinds: Vec<ColumnKindTag>,
297}
298
299impl EncodedDataset {
300 pub fn column_map(&self) -> HashMap<String, usize> {
301 self.headers
302 .iter()
303 .enumerate()
304 .map(|(index, header)| (header.clone(), index))
305 .collect()
306 }
307
308 pub fn feature_ranges(&self) -> Vec<(f64, f64)> {
314 self.values
321 .axis_iter(Axis(1))
322 .into_par_iter()
323 .map(|col| {
324 let (lo, hi) =
325 col.iter()
326 .fold((f64::INFINITY, f64::NEG_INFINITY), |(lo, hi), &v| {
327 if v.is_finite() {
328 (lo.min(v), hi.max(v))
329 } else {
330 (lo, hi)
331 }
332 });
333 if !lo.is_finite() || !hi.is_finite() {
334 (0.0, 0.0)
335 } else {
336 (lo, hi)
337 }
338 })
339 .collect()
340 }
341}
342
343fn shared_prefix(a: &str, b: &str) -> usize {
344 a.chars()
345 .zip(b.chars())
346 .take_while(|(ca, cb)| ca == cb)
347 .count()
348}
349
350#[derive(Clone, Copy, Debug, Eq, PartialEq)]
355enum DataFormat {
356 Csv,
357 Tsv,
358 Parquet,
359}
360
361fn detect_format(path: &Path) -> Result<DataFormat, DataError> {
362 let ext = path
363 .extension()
364 .and_then(|s| s.to_str())
365 .unwrap_or_default()
366 .to_ascii_lowercase();
367 match ext.as_str() {
368 "csv" => Ok(DataFormat::Csv),
369 "tsv" | "txt" | "tab" => Ok(DataFormat::Tsv),
370 "parquet" | "pq" | "pqt" => Ok(DataFormat::Parquet),
371 other => Err(DataError::ParseError {
372 reason: format!(
373 "unsupported data file extension '.{other}'; expected csv, tsv, txt, parquet, or pq: '{}'",
374 path.display()
375 ),
376 }),
377 }
378}
379
380pub fn load_dataset_projected(
385 path: &Path,
386 requested_columns: &[String],
387) -> Result<EncodedDataset, DataError> {
388 load_dataset_projected_with_categorical_roles(path, requested_columns, &HashSet::new())
389}
390
391pub fn load_dataset_projected_with_categorical_roles(
413 path: &Path,
414 requested_columns: &[String],
415 categorical_roles: &HashSet<&str>,
416) -> Result<EncodedDataset, DataError> {
417 match detect_format(path)? {
418 DataFormat::Csv => {
419 load_delimited_inferred(path, b',', requested_columns, categorical_roles)
420 }
421 DataFormat::Tsv => {
422 load_delimited_inferred(path, b'\t', requested_columns, categorical_roles)
423 }
424 DataFormat::Parquet => load_parquet_inferred(path, requested_columns, categorical_roles),
425 }
426}
427
428pub fn load_datasetwith_schema_projected(
429 path: &Path,
430 schema: &DataSchema,
431 unseen_policy: UnseenCategoryPolicy,
432 requested_columns: &[String],
433) -> Result<EncodedDataset, DataError> {
434 match detect_format(path)? {
435 DataFormat::Csv => {
436 load_delimited_with_schema(path, b',', schema, unseen_policy, requested_columns)
437 }
438 DataFormat::Tsv => {
439 load_delimited_with_schema(path, b'\t', schema, unseen_policy, requested_columns)
440 }
441 DataFormat::Parquet => {
442 load_parquet_with_schema(path, schema, unseen_policy, requested_columns)
443 }
444 }
445}
446
447pub fn load_csvwith_inferred_schema(path: &Path) -> Result<EncodedDataset, DataError> {
452 load_delimited_inferred(path, b',', &[], &HashSet::new())
453}
454
455const SCHEMA_SAMPLE_ROWS: usize = 1024;
461
462pub const CATEGORICAL_CELL_SENTINEL: char = '\u{0}';
472
473pub fn strip_categorical_sentinel(cell: &str) -> (&str, bool) {
476 match cell.strip_prefix(CATEGORICAL_CELL_SENTINEL) {
477 Some(rest) => (rest, true),
478 None => (cell, false),
479 }
480}
481
482fn resolve_requested_columns(
483 all_headers: &[String],
484 requested_columns: &[String],
485) -> Result<Vec<usize>, DataError> {
486 if requested_columns.is_empty() {
487 return Ok((0..all_headers.len()).collect());
488 }
489
490 let requested_set: HashSet<&str> = requested_columns.iter().map(String::as_str).collect();
491 let mut selected = Vec::with_capacity(requested_set.len());
492 for (idx, name) in all_headers.iter().enumerate() {
493 if requested_set.contains(name.as_str()) {
494 selected.push(idx);
495 }
496 }
497
498 if selected.len() != requested_set.len() {
499 let available_map: HashMap<String, usize> = all_headers
500 .iter()
501 .enumerate()
502 .map(|(index, header)| (header.clone(), index))
503 .collect();
504 let missing = requested_columns
505 .iter()
506 .filter(|name| !available_map.contains_key(name.as_str()))
507 .map(|name| {
508 DataError::column_not_found(&available_map, name, Some("requested")).to_string()
509 })
510 .collect::<Vec<_>>();
511 return Err(DataError::SchemaMismatch {
512 reason: missing.join("; "),
513 });
514 }
515
516 Ok(selected)
517}
518
519fn projected_headers(all_headers: &[String], selected_indices: &[usize]) -> Vec<String> {
520 selected_indices
521 .iter()
522 .map(|&idx| all_headers[idx].clone())
523 .collect()
524}
525
526fn load_delimited_inferred(
527 path: &Path,
528 delimiter: u8,
529 requested_columns: &[String],
530 categorical_roles: &HashSet<&str>,
531) -> Result<EncodedDataset, DataError> {
532 let t_open = std::time::Instant::now();
533 let mut rdr = ReaderBuilder::new()
534 .has_headers(true)
535 .delimiter(delimiter)
536 .from_path(path)
537 .map_err(|e| DataError::ParseError {
538 reason: format!("failed to open '{}': {e}", path.display()),
539 })?;
540
541 let all_headers: Vec<String> = rdr
542 .headers()
543 .map_err(|e| DataError::ParseError {
544 reason: format!("failed to read headers: {e}"),
545 })?
546 .iter()
547 .map(|s| s.trim().to_string())
548 .collect();
549 if all_headers.is_empty() {
550 return Err(DataError::EmptyInput {
551 reason: "file has no headers".to_string(),
552 });
553 }
554 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
555 let headers = projected_headers(&all_headers, &selected_indices);
556 let p = headers.len();
557 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
558 if open_ms > 100.0 {
559 log::info!(
560 "[DATA-LOAD] delim_open+headers | n_headers={} | n_proj={} | {:.1}ms",
561 all_headers.len(),
562 p,
563 open_ms
564 );
565 }
566
567 let mut raw_fields = Vec::<String>::new();
575 let mut total_rows: usize = 0;
576 let mut stream_error: Option<DataError> = None;
577
578 let t_stream = std::time::Instant::now();
579 let mut record = StringRecord::new();
580 while rdr
581 .read_record(&mut record)
582 .map_err(|e| DataError::ParseError {
583 reason: format!("failed reading row: {e}"),
584 })?
585 {
586 if record.len() != all_headers.len() {
587 stream_error = Some(DataError::SchemaMismatch {
588 reason: format!(
589 "row width mismatch at row {}: got {} fields, expected {}",
590 total_rows + 1,
591 record.len(),
592 all_headers.len()
593 ),
594 });
595 break;
596 }
597 total_rows += 1;
598
599 for &selected_idx in &selected_indices {
600 let raw = record.get(selected_idx).unwrap().trim();
601 raw_fields.push(raw.to_string());
602 }
603 }
604
605 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
606 if stream_ms > 100.0 {
607 log::info!(
608 "[DATA-LOAD] delim_stream | n_rows={} | n_cols={} | {:.1}ms",
609 total_rows,
610 p,
611 stream_ms
612 );
613 }
614
615 if total_rows == 0 {
616 if let Some(err) = stream_error {
617 return Err(err);
618 }
619 return Err(DataError::EmptyInput {
620 reason: "file has no rows".to_string(),
621 });
622 }
623
624 let t_schema = std::time::Instant::now();
625 let sample_count = total_rows.min(SCHEMA_SAMPLE_ROWS);
626 let inferred_columns = (0..p)
627 .into_par_iter()
628 .map(|j| {
629 infer_delimited_column(
630 &raw_fields,
631 total_rows,
632 p,
633 j,
634 &headers[j],
635 sample_count,
636 categorical_roles.contains(headers[j].as_str()),
637 )
638 })
639 .collect::<Vec<_>>();
640
641 let first_error = inferred_columns
642 .iter()
643 .filter_map(|result| result.as_ref().err())
644 .min_by_key(|err| (err.row, err.col));
645 if let Some(err) = first_error {
646 return Err(err.error.clone());
647 }
648 if let Some(err) = stream_error {
649 return Err(err);
650 }
651
652 let inferred_columns = inferred_columns
653 .into_iter()
654 .map(Result::unwrap)
655 .collect::<Vec<_>>();
656
657 let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
659 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
660 for (j, inferred) in inferred_columns.iter().enumerate() {
661 column_kinds.push(inferred.kind);
662 schema_cols.push(SchemaColumn {
663 name: headers[j].clone(),
664 kind: inferred.kind,
665 levels: if matches!(inferred.kind, ColumnKindTag::Categorical) {
666 inferred.levels.clone()
667 } else {
668 Vec::new()
669 },
670 });
671 }
672 let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
673 if schema_ms > 100.0 {
674 let n_cat = column_kinds
675 .iter()
676 .filter(|k| matches!(k, ColumnKindTag::Categorical))
677 .count();
678 log::info!(
679 "[DATA-LOAD] delim_convert+infer | n_cols={} | n_cat={} | {:.1}ms",
680 p,
681 n_cat,
682 schema_ms
683 );
684 }
685
686 let t_assemble = std::time::Instant::now();
687 let mut values = Array2::<f64>::zeros((total_rows, p));
689 values
690 .axis_iter_mut(Axis(1))
691 .into_par_iter()
692 .zip(inferred_columns.par_iter())
693 .for_each(|(mut out_col, inferred)| {
694 for (dst, &src) in out_col.iter_mut().zip(inferred.values.iter()) {
695 *dst = src;
696 }
697 });
698 let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
699 if assemble_ms > 100.0 {
700 log::info!(
701 "[DATA-LOAD] delim_assemble_array2 | n_rows={} | n_cols={} | {:.1}ms",
702 total_rows,
703 p,
704 assemble_ms
705 );
706 }
707
708 let schema = DataSchema {
709 columns: schema_cols,
710 };
711 Ok(EncodedDataset {
712 headers,
713 values,
714 schema,
715 column_kinds,
716 })
717}
718
719struct InferredDelimitedColumn {
720 values: Vec<f64>,
721 kind: ColumnKindTag,
722 levels: Vec<String>,
723}
724
725#[derive(Debug)]
726struct DelimitedInferenceError {
727 row: usize,
728 col: usize,
729 error: DataError,
730}
731
732fn infer_delimited_column(
733 raw_fields: &[String],
734 total_rows: usize,
735 n_cols: usize,
736 col: usize,
737 header: &str,
738 sample_count: usize,
739 force_categorical: bool,
740) -> Result<InferredDelimitedColumn, DelimitedInferenceError> {
741 let mut values = Vec::<f64>::with_capacity(total_rows);
743 let mut all_numeric = true;
744 let mut all_binary = true;
745 let mut level_index = HashMap::<String, usize>::new();
746 let mut levels = Vec::<String>::new();
747
748 let non_finite_err = |row_idx: usize| DelimitedInferenceError {
752 row: row_idx + 1,
753 col,
754 error: DataError::InvalidValue {
755 reason: format!(
756 "non-finite value at row {}, column '{}'",
757 row_idx + 1,
758 header
759 ),
760 },
761 };
762
763 for row_idx in 0..total_rows {
764 let raw = raw_fields[row_idx * n_cols + col].as_str();
765 if raw.is_empty() {
766 return Err(DelimitedInferenceError {
767 row: row_idx + 1,
768 col,
769 error: DataError::EmptyInput {
770 reason: format!("empty field at row {}, column '{}'", row_idx + 1, header),
771 },
772 });
773 }
774
775 if row_idx < sample_count {
777 if let Ok(v) = raw.parse::<f64>() {
778 if !v.is_finite() {
779 return Err(non_finite_err(row_idx));
780 }
781 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
782 all_binary = false;
783 }
784 values.push(v);
785 } else {
786 all_numeric = false;
787 all_binary = false;
788 level_index.entry(raw.to_string()).or_insert_with(|| {
789 let idx = levels.len();
790 levels.push(raw.to_string());
791 idx
792 });
793 values.push(f64::NAN);
797 }
798 } else if let Ok(v) = raw.parse::<f64>() {
799 if !v.is_finite() {
803 return Err(non_finite_err(row_idx));
804 }
805 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
806 all_binary = false;
807 }
808 values.push(v);
809 } else {
810 all_numeric = false;
811 all_binary = false;
812 let idx = *level_index.entry(raw.to_string()).or_insert_with(|| {
813 let new_idx = levels.len();
814 levels.push(raw.to_string());
815 new_idx
816 });
817 values.push(idx as f64);
818 }
819 }
820
821 let kind = if force_categorical {
830 ColumnKindTag::Categorical
831 } else if all_numeric {
832 if all_binary {
833 ColumnKindTag::Binary
834 } else {
835 ColumnKindTag::Continuous
836 }
837 } else {
838 ColumnKindTag::Categorical
839 };
840
841 if matches!(kind, ColumnKindTag::Categorical) {
842 for row_idx in 0..total_rows {
863 let raw = raw_fields[row_idx * n_cols + col].as_str();
864 level_index.entry(raw.to_string()).or_insert_with(|| {
865 let new_idx = levels.len();
866 levels.push(raw.to_string());
867 new_idx
868 });
869 }
870 sort_levels_canonical(&mut levels);
871 level_index.clear();
872 for (idx, level) in levels.iter().enumerate() {
873 level_index.insert(level.clone(), idx);
874 }
875 for row_idx in 0..total_rows {
876 let raw = raw_fields[row_idx * n_cols + col].as_str();
877 values[row_idx] = level_index[raw] as f64;
878 }
879 }
880
881 for (row_idx, &v) in values.iter().enumerate() {
882 if !v.is_finite() {
883 return Err(non_finite_err(row_idx));
884 }
885 }
886
887 Ok(InferredDelimitedColumn {
888 values,
889 kind,
890 levels,
891 })
892}
893
894fn load_delimited_with_schema(
895 path: &Path,
896 delimiter: u8,
897 schema: &DataSchema,
898 unseen_policy: UnseenCategoryPolicy,
899 requested_columns: &[String],
900) -> Result<EncodedDataset, DataError> {
901 let t_open = std::time::Instant::now();
902 let mut rdr = ReaderBuilder::new()
903 .has_headers(true)
904 .delimiter(delimiter)
905 .from_path(path)
906 .map_err(|e| DataError::ParseError {
907 reason: format!("failed to open '{}': {e}", path.display()),
908 })?;
909
910 let all_headers: Vec<String> = rdr
911 .headers()
912 .map_err(|e| DataError::ParseError {
913 reason: format!("failed to read headers: {e}"),
914 })?
915 .iter()
916 .map(|s| s.trim().to_string())
917 .collect();
918 if all_headers.is_empty() {
919 return Err(DataError::EmptyInput {
920 reason: "file has no headers".to_string(),
921 });
922 }
923 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
924 let headers = projected_headers(&all_headers, &selected_indices);
925 let p = headers.len();
926 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
927 if open_ms > 100.0 {
928 log::info!(
929 "[DATA-LOAD] delim_schema_open+headers | n_headers={} | n_proj={} | {:.1}ms",
930 all_headers.len(),
931 p,
932 open_ms
933 );
934 }
935
936 let schema_byname: HashMap<&str, &SchemaColumn> = schema
938 .columns
939 .iter()
940 .map(|c| (c.name.as_str(), c))
941 .collect();
942
943 let mut col_meta = Vec::<ColMeta>::with_capacity(p);
944 for name in &headers {
945 if let Some(sc) = schema_byname.get(name.as_str()) {
946 let level_map = if matches!(sc.kind, ColumnKindTag::Categorical) {
947 Some(
948 sc.levels
949 .iter()
950 .enumerate()
951 .map(|(idx, v)| (v.clone(), idx as f64))
952 .collect::<HashMap<_, _>>(),
953 )
954 } else {
955 None
956 };
957 col_meta.push(ColMeta {
958 kind: sc.kind,
959 level_map,
960 schema_col: (*sc).clone(),
961 });
962 } else {
963 col_meta.push(ColMeta {
965 kind: ColumnKindTag::Continuous, level_map: None,
967 schema_col: SchemaColumn {
968 name: name.clone(),
969 kind: ColumnKindTag::Continuous,
970 levels: Vec::new(),
971 },
972 });
973 }
974 }
975
976 let needs_inference: Vec<bool> = headers
978 .iter()
979 .map(|h| !schema_byname.contains_key(h.as_str()))
980 .collect();
981
982 let mut col_vecs: Vec<Vec<f64>> = vec![Vec::new(); p];
984 let mut infer_all_numeric: Vec<bool> = vec![true; p];
986 let mut infer_all_binary: Vec<bool> = vec![true; p];
987 let mut infer_level_index: Vec<HashMap<String, usize>> = vec![HashMap::new(); p];
988 let mut infer_levels: Vec<Vec<String>> = vec![Vec::new(); p];
989 let mut infer_strings: Vec<Vec<(usize, String)>> = vec![Vec::new(); p]; let mut total_rows: usize = 0;
992 let t_stream = std::time::Instant::now();
993 let mut record = StringRecord::new();
994 while rdr
995 .read_record(&mut record)
996 .map_err(|e| DataError::ParseError {
997 reason: format!("failed reading row: {e}"),
998 })?
999 {
1000 if record.len() != all_headers.len() {
1001 return Err(DataError::SchemaMismatch {
1002 reason: format!(
1003 "row width mismatch at row {}: got {} fields, expected {}",
1004 total_rows + 1,
1005 record.len(),
1006 all_headers.len()
1007 ),
1008 });
1009 }
1010 total_rows += 1;
1011
1012 for j in 0..p {
1013 let raw = record.get(selected_indices[j]).unwrap().trim();
1014 if raw.is_empty() {
1015 return Err(DataError::EmptyInput {
1016 reason: format!(
1017 "empty field at row {}, column '{}'",
1018 total_rows, &headers[j]
1019 ),
1020 });
1021 }
1022
1023 if needs_inference[j] {
1024 if let Ok(v) = raw.parse::<f64>() {
1026 if !v.is_finite() {
1027 return Err(DataError::InvalidValue {
1028 reason: format!(
1029 "non-finite value at row {}, column '{}'",
1030 total_rows, &headers[j]
1031 ),
1032 });
1033 }
1034 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
1035 infer_all_binary[j] = false;
1036 }
1037 col_vecs[j].push(v);
1038 infer_strings[j].push((total_rows - 1, raw.to_string()));
1047 } else {
1048 infer_all_numeric[j] = false;
1049 infer_all_binary[j] = false;
1050 let levels_ref = &mut infer_levels[j];
1051 infer_level_index[j]
1052 .entry(raw.to_string())
1053 .or_insert_with(|| {
1054 let idx = levels_ref.len();
1055 levels_ref.push(raw.to_string());
1056 idx
1057 });
1058 infer_strings[j].push((total_rows - 1, raw.to_string()));
1059 col_vecs[j].push(f64::NAN); }
1061 } else {
1062 let val = parse_cell_with_schema(
1064 raw,
1065 &col_meta[j],
1066 total_rows,
1067 &headers[j],
1068 &unseen_policy,
1069 )?;
1070 col_vecs[j].push(val);
1071 }
1072 }
1073 }
1074
1075 let stream_ms = t_stream.elapsed().as_secs_f64() * 1000.0;
1076 if stream_ms > 100.0 {
1077 let n_inf = needs_inference.iter().filter(|x| **x).count();
1078 log::info!(
1079 "[DATA-LOAD] delim_schema_stream | n_rows={} | n_cols={} | n_inf={} | {:.1}ms",
1080 total_rows,
1081 p,
1082 n_inf,
1083 stream_ms
1084 );
1085 }
1086
1087 if total_rows == 0 {
1088 return Err(DataError::EmptyInput {
1089 reason: "file has no rows".to_string(),
1090 });
1091 }
1092
1093 let t_finalize = std::time::Instant::now();
1094 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
1096 for j in 0..p {
1097 if needs_inference[j] {
1098 let kind = if infer_all_numeric[j] {
1099 if infer_all_binary[j] {
1100 ColumnKindTag::Binary
1101 } else {
1102 ColumnKindTag::Continuous
1103 }
1104 } else {
1105 ColumnKindTag::Categorical
1106 };
1107 col_meta[j].kind = kind;
1108 col_meta[j].schema_col.kind = kind;
1109 if matches!(kind, ColumnKindTag::Categorical) {
1110 for (_, raw) in &infer_strings[j] {
1123 let levels_ref = &mut infer_levels[j];
1124 infer_level_index[j].entry(raw.clone()).or_insert_with(|| {
1125 let new_idx = levels_ref.len();
1126 levels_ref.push(raw.clone());
1127 new_idx
1128 });
1129 }
1130 infer_levels[j].sort();
1131 infer_level_index[j].clear();
1132 for (idx, level) in infer_levels[j].iter().enumerate() {
1133 infer_level_index[j].insert(level.clone(), idx);
1134 }
1135 for (row_idx, raw) in &infer_strings[j] {
1136 col_vecs[j][*row_idx] = infer_level_index[j][raw] as f64;
1137 }
1138 col_meta[j].schema_col.levels = infer_levels[j].clone();
1139 }
1140 }
1141 column_kinds.push(col_meta[j].kind);
1142 }
1143 let finalize_ms = t_finalize.elapsed().as_secs_f64() * 1000.0;
1144 if finalize_ms > 100.0 {
1145 log::info!(
1146 "[DATA-LOAD] delim_schema_finalize | n_cols={} | {:.1}ms",
1147 p,
1148 finalize_ms
1149 );
1150 }
1151
1152 let t_assemble = std::time::Instant::now();
1153 let mut values = Array2::<f64>::zeros((total_rows, p));
1159 let assemble_err: Option<DataError> = values
1160 .axis_iter_mut(Axis(1))
1161 .into_par_iter()
1162 .zip(col_vecs.par_iter())
1163 .zip(headers.par_iter())
1164 .map(|((mut out_col, col_vec), header)| {
1165 for (i, &v) in col_vec.iter().enumerate() {
1166 if !v.is_finite() {
1167 return Some(DataError::InvalidValue {
1168 reason: format!("non-finite value at row {}, column '{}'", i + 1, header),
1169 });
1170 }
1171 out_col[i] = v;
1172 }
1173 None
1174 })
1175 .reduce(|| None, |a, b| a.or(b));
1176 if let Some(e) = assemble_err {
1177 return Err(e);
1178 }
1179 let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
1180 if assemble_ms > 100.0 {
1181 log::info!(
1182 "[DATA-LOAD] delim_schema_assemble | n_rows={} | n_cols={} | {:.1}ms",
1183 total_rows,
1184 p,
1185 assemble_ms
1186 );
1187 }
1188
1189 let schema_out = DataSchema {
1190 columns: col_meta.into_iter().map(|m| m.schema_col).collect(),
1191 };
1192 Ok(EncodedDataset {
1193 headers,
1194 values,
1195 schema: schema_out,
1196 column_kinds,
1197 })
1198}
1199
1200fn parse_cell_with_schema(
1201 raw: &str,
1202 meta: &ColMeta,
1203 row: usize,
1204 col_name: &str,
1205 unseen_policy: &UnseenCategoryPolicy,
1206) -> Result<f64, DataError> {
1207 let val = match meta.kind {
1208 ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
1209 DataError::SchemaMismatch {
1210 reason: format!(
1211 "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
1212 col_name, row, raw, err
1213 ),
1214 }
1215 })?,
1216 ColumnKindTag::Binary => {
1217 let v = raw
1218 .parse::<f64>()
1219 .map_err(|err| DataError::SchemaMismatch {
1220 reason: format!(
1221 "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
1222 col_name, row, raw, err
1223 ),
1224 })?;
1225 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
1226 return Err(DataError::SchemaMismatch {
1227 reason: format!(
1228 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
1229 col_name, row, v
1230 ),
1231 });
1232 }
1233 v
1234 }
1235 ColumnKindTag::Categorical => {
1236 let map = meta
1237 .level_map
1238 .as_ref()
1239 .ok_or_else(|| DataError::EncodingFailure {
1240 reason: "internal categorical schema map missing".to_string(),
1241 })?;
1242 match map.get(raw) {
1243 Some(v) => *v,
1244 None => unseen_policy
1245 .unseen_code_for(col_name, meta.schema_col.levels.len())
1246 .ok_or_else(|| DataError::SchemaMismatch {
1247 reason: format!(
1248 "unseen level '{}' in categorical column '{}' at row {}",
1249 raw, col_name, row
1250 ),
1251 })?,
1252 }
1253 }
1254 };
1255 if !val.is_finite() {
1256 return Err(DataError::InvalidValue {
1257 reason: format!("non-finite value at row {}, column '{}'", row, col_name),
1258 });
1259 }
1260 Ok(val)
1261}
1262
1263struct ColMeta {
1266 kind: ColumnKindTag,
1267 level_map: Option<HashMap<String, f64>>,
1268 schema_col: SchemaColumn,
1269}
1270
1271enum ParquetBatchColumn {
1276 Numeric(Vec<f64>),
1277 Strings(Vec<String>),
1278}
1279
1280fn parquet_field_is_string(dt: &arrow::datatypes::DataType) -> bool {
1289 use arrow::datatypes::DataType;
1290 match dt {
1291 DataType::Utf8 | DataType::LargeUtf8 => true,
1292 DataType::Dictionary(_, value_type) => parquet_field_is_string(value_type),
1293 _ => false,
1294 }
1295}
1296
1297fn decode_parquet_batch_column(
1298 col: &dyn arrow::array::Array,
1299 n_rows: usize,
1300 base_row: usize,
1301 header: &str,
1302 is_string_col: bool,
1303) -> Result<ParquetBatchColumn, DataError> {
1304 use arrow::array::{
1305 Array as ArrowArray, BooleanArray, Float32Array, Float64Array, Int8Array, Int16Array,
1306 Int32Array, Int64Array, LargeStringArray, StringArray, UInt8Array, UInt16Array,
1307 UInt32Array, UInt64Array,
1308 };
1309 use arrow::datatypes::DataType;
1310
1311 if col.null_count() > 0 {
1312 for i in 0..n_rows {
1313 if col.is_null(i) {
1314 return Err(DataError::InvalidValue {
1315 reason: format!(
1316 "null value at row {}, column '{}'",
1317 base_row + i + 1,
1318 header
1319 ),
1320 });
1321 }
1322 }
1323 }
1324
1325 if is_string_col {
1326 if let Some(arr) = col.as_any().downcast_ref::<StringArray>() {
1327 return Ok(ParquetBatchColumn::Strings(
1328 (0..n_rows).map(|i| arr.value(i).to_string()).collect(),
1329 ));
1330 }
1331 if let Some(arr) = col.as_any().downcast_ref::<LargeStringArray>() {
1332 return Ok(ParquetBatchColumn::Strings(
1333 (0..n_rows).map(|i| arr.value(i).to_string()).collect(),
1334 ));
1335 }
1336
1337 let casted =
1341 arrow::compute::cast(col, &DataType::Utf8).map_err(|e| DataError::ParseError {
1342 reason: format!("failed to cast column '{}' to string: {e}", header),
1343 })?;
1344 let arr = casted
1345 .as_any()
1346 .downcast_ref::<StringArray>()
1347 .ok_or_else(|| DataError::EncodingFailure {
1348 reason: format!("column '{}' could not be read as string after cast", header),
1349 })?;
1350 return Ok(ParquetBatchColumn::Strings(
1351 (0..n_rows).map(|i| arr.value(i).to_string()).collect(),
1352 ));
1353 }
1354
1355 let decoded_col;
1362 let col: &dyn arrow::array::Array = if let DataType::Dictionary(_, value_type) = col.data_type()
1363 {
1364 decoded_col = arrow::compute::cast(col, value_type).map_err(|e| DataError::ParseError {
1365 reason: format!(
1366 "failed to decode dictionary-encoded numeric column '{}': {e}",
1367 header
1368 ),
1369 })?;
1370 decoded_col.as_ref()
1371 } else {
1372 col
1373 };
1374
1375 let mut values = Vec::with_capacity(n_rows);
1376 match col.data_type() {
1377 DataType::Float64 => {
1378 let arr = col.as_any().downcast_ref::<Float64Array>().unwrap();
1379 values.extend(arr.values().iter().copied());
1380 }
1381 DataType::Float32 => {
1382 let arr = col.as_any().downcast_ref::<Float32Array>().unwrap();
1383 values.extend(arr.values().iter().map(|&v| v as f64));
1384 }
1385 DataType::Int64 => {
1386 let arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
1387 values.extend(arr.values().iter().map(|&v| v as f64));
1388 }
1389 DataType::Int32 => {
1390 let arr = col.as_any().downcast_ref::<Int32Array>().unwrap();
1391 values.extend(arr.values().iter().map(|&v| v as f64));
1392 }
1393 DataType::Int16 => {
1394 let arr = col.as_any().downcast_ref::<Int16Array>().unwrap();
1395 values.extend(arr.values().iter().map(|&v| v as f64));
1396 }
1397 DataType::Int8 => {
1398 let arr = col.as_any().downcast_ref::<Int8Array>().unwrap();
1399 values.extend(arr.values().iter().map(|&v| v as f64));
1400 }
1401 DataType::UInt64 => {
1402 let arr = col.as_any().downcast_ref::<UInt64Array>().unwrap();
1403 values.extend(arr.values().iter().map(|&v| v as f64));
1404 }
1405 DataType::UInt32 => {
1406 let arr = col.as_any().downcast_ref::<UInt32Array>().unwrap();
1407 values.extend(arr.values().iter().map(|&v| v as f64));
1408 }
1409 DataType::UInt16 => {
1410 let arr = col.as_any().downcast_ref::<UInt16Array>().unwrap();
1411 values.extend(arr.values().iter().map(|&v| v as f64));
1412 }
1413 DataType::UInt8 => {
1414 let arr = col.as_any().downcast_ref::<UInt8Array>().unwrap();
1415 values.extend(arr.values().iter().map(|&v| v as f64));
1416 }
1417 DataType::Boolean => {
1418 let arr = col.as_any().downcast_ref::<BooleanArray>().unwrap();
1419 values.extend((0..n_rows).map(|i| if arr.value(i) { 1.0 } else { 0.0 }));
1420 }
1421 other => {
1422 return Err(DataError::InvalidValue {
1423 reason: format!(
1424 "unsupported parquet column type {:?} for column '{}'",
1425 other, header
1426 ),
1427 });
1428 }
1429 }
1430
1431 if let Some(i) = values.iter().position(|v| !v.is_finite()) {
1432 return Err(DataError::InvalidValue {
1433 reason: format!(
1434 "non-finite value at row {}, column '{}'",
1435 base_row + i + 1,
1436 header
1437 ),
1438 });
1439 }
1440
1441 Ok(ParquetBatchColumn::Numeric(values))
1442}
1443
1444fn load_parquet_inferred(
1445 path: &Path,
1446 requested_columns: &[String],
1447 categorical_roles: &HashSet<&str>,
1448) -> Result<EncodedDataset, DataError> {
1449 use parquet::arrow::{ProjectionMask, arrow_reader::ParquetRecordBatchReaderBuilder};
1450 use rayon::prelude::*;
1451 use std::fs::File;
1452
1453 let t_open = std::time::Instant::now();
1454 let file = File::open(path).map_err(|e| DataError::ParseError {
1455 reason: format!("failed to open parquet '{}': {e}", path.display()),
1456 })?;
1457 let builder =
1458 ParquetRecordBatchReaderBuilder::try_new(file).map_err(|e| DataError::ParseError {
1459 reason: format!("failed to read parquet metadata '{}': {e}", path.display()),
1460 })?;
1461
1462 let full_schema = builder.schema().clone();
1463 let all_headers: Vec<String> = full_schema
1464 .fields()
1465 .iter()
1466 .map(|f| f.name().clone())
1467 .collect();
1468 if all_headers.is_empty() {
1469 return Err(DataError::EmptyInput {
1470 reason: "parquet file has no columns".to_string(),
1471 });
1472 }
1473 let selected_indices = resolve_requested_columns(&all_headers, requested_columns)?;
1474 let headers = projected_headers(&all_headers, &selected_indices);
1475 let selected_fields = selected_indices
1476 .iter()
1477 .map(|&idx| full_schema.fields()[idx].clone())
1478 .collect::<Vec<_>>();
1479 let projection =
1480 ProjectionMask::roots(builder.parquet_schema(), selected_indices.iter().copied());
1481 let reader =
1482 builder
1483 .with_projection(projection)
1484 .build()
1485 .map_err(|e| DataError::ParseError {
1486 reason: format!("failed to build parquet reader: {e}"),
1487 })?;
1488 let p = headers.len();
1489 let open_ms = t_open.elapsed().as_secs_f64() * 1000.0;
1490 if open_ms > 100.0 {
1491 log::info!(
1492 "[DATA-LOAD] parquet_open+meta | n_headers={} | n_proj={} | {:.1}ms",
1493 all_headers.len(),
1494 p,
1495 open_ms
1496 );
1497 }
1498
1499 let t_batches = std::time::Instant::now();
1500 let mut col_vecs: Vec<Vec<f64>> = vec![Vec::new(); p];
1502 let mut string_cols: Vec<Option<Vec<String>>> = (0..p).map(|_| None).collect();
1504 let mut is_string_col: Vec<bool> = vec![false; p];
1505
1506 for (j, field) in selected_fields.iter().enumerate() {
1507 if parquet_field_is_string(field.data_type()) {
1511 is_string_col[j] = true;
1512 string_cols[j] = Some(Vec::new());
1513 }
1514 }
1515
1516 let mut rows_seen = 0usize;
1517 for batch_result in reader {
1518 let batch = batch_result.map_err(|e| DataError::ParseError {
1519 reason: format!("failed to read parquet record batch: {e}"),
1520 })?;
1521 let n_rows = batch.num_rows();
1522
1523 let decoded_columns: Vec<Result<ParquetBatchColumn, DataError>> = (0..p)
1524 .into_par_iter()
1525 .map(|j| {
1526 decode_parquet_batch_column(
1527 batch.column(j).as_ref(),
1528 n_rows,
1529 rows_seen,
1530 &headers[j],
1531 is_string_col[j],
1532 )
1533 })
1534 .collect();
1535
1536 for (j, decoded) in decoded_columns.into_iter().enumerate() {
1537 match decoded? {
1538 ParquetBatchColumn::Strings(mut strings) => {
1539 assert!(is_string_col[j]);
1540 string_cols[j].as_mut().unwrap().append(&mut strings);
1541 let new_len = col_vecs[j].len() + n_rows;
1542 col_vecs[j].resize(new_len, f64::NAN);
1543 }
1544 ParquetBatchColumn::Numeric(mut values) => {
1545 assert!(!is_string_col[j]);
1546 col_vecs[j].append(&mut values);
1547 }
1548 }
1549 }
1550 rows_seen += n_rows;
1551 }
1552
1553 let total_rows = col_vecs[0].len();
1554 let batches_ms = t_batches.elapsed().as_secs_f64() * 1000.0;
1555 if batches_ms > 100.0 {
1556 log::info!(
1557 "[DATA-LOAD] parquet_batches_decode | n_rows={} | n_cols={} | {:.1}ms",
1558 total_rows,
1559 p,
1560 batches_ms
1561 );
1562 }
1563 if total_rows == 0 {
1564 return Err(DataError::EmptyInput {
1565 reason: "parquet file has no rows".to_string(),
1566 });
1567 }
1568
1569 let t_schema = std::time::Instant::now();
1570 let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
1572 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
1573
1574 let finalized_columns: Vec<(Vec<f64>, ColumnKindTag, SchemaColumn)> = col_vecs
1575 .into_par_iter()
1576 .zip(string_cols.into_par_iter())
1577 .zip(is_string_col.into_par_iter())
1578 .zip(headers.par_iter())
1579 .map(|(((mut col_values, strings), is_string), header)| {
1580 if is_string {
1581 let strings = strings.expect("string column storage missing");
1585 let mut level_index: HashMap<String, usize> = HashMap::new();
1586 let mut levels_vec: Vec<String> = Vec::new();
1587 for s in &strings {
1588 level_index.entry(s.clone()).or_insert_with(|| {
1589 let idx = levels_vec.len();
1590 levels_vec.push(s.clone());
1591 idx
1592 });
1593 }
1594 for (i, s) in strings.iter().enumerate() {
1595 col_values[i] = *level_index.get(s.as_str()).unwrap() as f64;
1596 }
1597 (
1598 col_values,
1599 ColumnKindTag::Categorical,
1600 SchemaColumn {
1601 name: header.clone(),
1602 kind: ColumnKindTag::Categorical,
1603 levels: levels_vec,
1604 },
1605 )
1606 } else if categorical_roles.contains(header.as_str()) {
1607 let labels: Vec<String> = col_values.iter().map(|v| v.to_string()).collect();
1615 let mut levels_vec: Vec<String> = Vec::new();
1616 let mut level_index: HashMap<String, usize> = HashMap::new();
1617 for label in &labels {
1618 level_index.entry(label.clone()).or_insert_with(|| {
1619 let idx = levels_vec.len();
1620 levels_vec.push(label.clone());
1621 idx
1622 });
1623 }
1624 levels_vec.sort();
1625 level_index.clear();
1626 for (idx, level) in levels_vec.iter().enumerate() {
1627 level_index.insert(level.clone(), idx);
1628 }
1629 for (i, label) in labels.iter().enumerate() {
1630 col_values[i] = level_index[label] as f64;
1631 }
1632 (
1633 col_values,
1634 ColumnKindTag::Categorical,
1635 SchemaColumn {
1636 name: header.clone(),
1637 kind: ColumnKindTag::Categorical,
1638 levels: levels_vec,
1639 },
1640 )
1641 } else {
1642 let all_binary = col_values
1644 .iter()
1645 .all(|&v| (v - 0.0).abs() < 1e-12 || (v - 1.0).abs() < 1e-12);
1646 let kind = if all_binary {
1647 ColumnKindTag::Binary
1648 } else {
1649 ColumnKindTag::Continuous
1650 };
1651 (
1652 col_values,
1653 kind,
1654 SchemaColumn {
1655 name: header.clone(),
1656 kind,
1657 levels: Vec::new(),
1658 },
1659 )
1660 }
1661 })
1662 .collect();
1663
1664 let mut col_vecs = Vec::with_capacity(p);
1665 for (col_values, kind, schema_col) in finalized_columns {
1666 col_vecs.push(col_values);
1667 column_kinds.push(kind);
1668 schema_cols.push(schema_col);
1669 }
1670 let schema_ms = t_schema.elapsed().as_secs_f64() * 1000.0;
1671 if schema_ms > 100.0 {
1672 let n_cat = column_kinds
1673 .iter()
1674 .filter(|k| matches!(k, ColumnKindTag::Categorical))
1675 .count();
1676 log::info!(
1677 "[DATA-LOAD] parquet_finalize_schema | n_cols={} | n_cat={} | {:.1}ms",
1678 p,
1679 n_cat,
1680 schema_ms
1681 );
1682 }
1683
1684 let t_assemble = std::time::Instant::now();
1685 let mut values = Array2::<f64>::zeros((total_rows, p));
1690 values
1691 .axis_iter_mut(Axis(1))
1692 .into_par_iter()
1693 .zip(col_vecs.par_iter())
1694 .for_each(|(mut out_col, src)| {
1695 for (dst, &v) in out_col.iter_mut().zip(src.iter()) {
1696 *dst = v;
1697 }
1698 });
1699 let assemble_ms = t_assemble.elapsed().as_secs_f64() * 1000.0;
1700 if assemble_ms > 100.0 {
1701 log::info!(
1702 "[DATA-LOAD] parquet_assemble_array2 | n_rows={} | n_cols={} | {:.1}ms",
1703 total_rows,
1704 p,
1705 assemble_ms
1706 );
1707 }
1708
1709 Ok(EncodedDataset {
1710 headers,
1711 values,
1712 schema: DataSchema {
1713 columns: schema_cols,
1714 },
1715 column_kinds,
1716 })
1717}
1718
1719fn load_parquet_with_schema(
1720 path: &Path,
1721 schema: &DataSchema,
1722 unseen_policy: UnseenCategoryPolicy,
1723 requested_columns: &[String],
1724) -> Result<EncodedDataset, DataError> {
1725 let inferred = load_parquet_inferred(path, requested_columns, &HashSet::new())?;
1729 let p = inferred.headers.len();
1730 let n = inferred.values.nrows();
1731
1732 let schema_byname: HashMap<&str, &SchemaColumn> = schema
1733 .columns
1734 .iter()
1735 .map(|c| (c.name.as_str(), c))
1736 .collect();
1737
1738 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
1739 let mut schema_cols = Vec::<SchemaColumn>::with_capacity(p);
1740 let mut values = inferred.values;
1741
1742 for j in 0..p {
1743 let name = &inferred.headers[j];
1744 if let Some(sc) = schema_byname.get(name.as_str()) {
1745 column_kinds.push(sc.kind);
1746 schema_cols.push((*sc).clone());
1747
1748 match sc.kind {
1749 ColumnKindTag::Continuous => {
1750 if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
1751 return Err(DataError::SchemaMismatch {
1752 reason: format!(
1753 "column '{}' is continuous in schema but parquet column is string/categorical",
1754 name
1755 ),
1756 });
1757 }
1758 }
1759 ColumnKindTag::Binary => {
1760 if matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
1761 return Err(DataError::SchemaMismatch {
1762 reason: format!(
1763 "column '{}' is binary in schema but parquet column is string/categorical",
1764 name
1765 ),
1766 });
1767 }
1768 if let Some(row) = values.column(j).iter().position(|value| {
1769 (*value - 0.0).abs() >= 1e-12 && (*value - 1.0).abs() >= 1e-12
1770 }) {
1771 return Err(DataError::SchemaMismatch {
1772 reason: format!(
1773 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
1774 name,
1775 row + 1,
1776 values[[row, j]]
1777 ),
1778 });
1779 }
1780 }
1781 ColumnKindTag::Categorical => {
1782 if !matches!(inferred.column_kinds[j], ColumnKindTag::Categorical) {
1783 return Err(DataError::SchemaMismatch {
1784 reason: format!(
1785 "column '{}' is categorical in schema but parquet column is numeric",
1786 name
1787 ),
1788 });
1789 }
1790 let inferred_col = &inferred.schema.columns[j];
1791 let schema_level_map: HashMap<&str, f64> = sc
1793 .levels
1794 .iter()
1795 .enumerate()
1796 .map(|(idx, v)| (v.as_str(), idx as f64))
1797 .collect();
1798 let inferred_to_schema: Vec<f64> = inferred_col
1799 .levels
1800 .iter()
1801 .map(|lv| {
1802 schema_level_map
1803 .get(lv.as_str())
1804 .copied()
1805 .or_else(|| unseen_policy.unseen_code_for(name, sc.levels.len()))
1806 .ok_or_else(|| DataError::SchemaMismatch {
1807 reason: format!(
1808 "unseen level '{}' in categorical column '{}'",
1809 lv, name
1810 ),
1811 })
1812 })
1813 .collect::<Result<Vec<_>, _>>()?;
1814 for i in 0..n {
1815 let old_code = values[[i, j]] as usize;
1816 if old_code >= inferred_to_schema.len() {
1817 let Some(unseen_code) =
1818 unseen_policy.unseen_code_for(name, sc.levels.len())
1819 else {
1820 return Err(DataError::SchemaMismatch {
1821 reason: format!(
1822 "unseen categorical code at row {}, column '{}'",
1823 i + 1,
1824 name
1825 ),
1826 });
1827 };
1828 values[[i, j]] = unseen_code;
1829 continue;
1830 }
1831 values[[i, j]] = inferred_to_schema[old_code];
1832 }
1833 }
1834 }
1835 } else {
1836 column_kinds.push(inferred.column_kinds[j]);
1838 schema_cols.push(inferred.schema.columns[j].clone());
1839 }
1840 }
1841
1842 Ok(EncodedDataset {
1843 headers: inferred.headers,
1844 values,
1845 schema: DataSchema {
1846 columns: schema_cols,
1847 },
1848 column_kinds,
1849 })
1850}
1851
1852pub fn encode_recordswith_inferred_schema(
1853 headers: Vec<String>,
1854 records: Vec<StringRecord>,
1855) -> Result<EncodedDataset, String> {
1856 if records.is_empty() {
1857 return Err(DataError::EmptyInput {
1858 reason: "table data cannot be empty".to_string(),
1859 }
1860 .into());
1861 }
1862 let schema_cols = headers
1868 .par_iter()
1869 .enumerate()
1870 .map(|(j, name)| infer_schema_column(name, &records, j).map_err(String::from))
1871 .collect::<Result<Vec<SchemaColumn>, String>>()?;
1872 let schema = DataSchema {
1873 columns: schema_cols,
1874 };
1875 encode_recordswith_schema(headers, records, &schema, UnseenCategoryPolicy::Error)
1876}
1877
1878pub fn encode_recordswith_schema(
1879 headers: Vec<String>,
1880 records: Vec<StringRecord>,
1881 schema: &DataSchema,
1882 unseen_policy: UnseenCategoryPolicy,
1883) -> Result<EncodedDataset, String> {
1884 let n = records.len();
1885 if n == 0 {
1886 return Err(DataError::EmptyInput {
1887 reason: "table data cannot be empty".to_string(),
1888 }
1889 .into());
1890 }
1891 let p = headers.len();
1892 if p == 0 {
1893 return Err(DataError::EmptyInput {
1894 reason: "table data must have at least one header column".to_string(),
1895 }
1896 .into());
1897 }
1898 for (i, rec) in records.iter().enumerate() {
1905 if rec.len() != p {
1906 return Err(DataError::SchemaMismatch {
1907 reason: format!(
1908 "row width mismatch at row {}: got {} fields, expected {} (one per header)",
1909 i + 1,
1910 rec.len(),
1911 p
1912 ),
1913 }
1914 .into());
1915 }
1916 }
1917 let schema_byname: HashMap<&str, &SchemaColumn> = schema
1918 .columns
1919 .iter()
1920 .map(|c| (c.name.as_str(), c))
1921 .collect();
1922
1923 let encoded_columns = headers
1929 .par_iter()
1930 .enumerate()
1931 .map(|(j, name)| {
1932 let inferred_for_extra;
1933 let col_schema = if let Some(s) = schema_byname.get(name.as_str()) {
1934 *s
1935 } else {
1936 inferred_for_extra =
1937 infer_schema_column(name, &records, j).map_err(String::from)?;
1938 &inferred_for_extra
1939 };
1940 let column = encode_one_column(name, &records, j, col_schema, &unseen_policy)?;
1941 Ok::<(ColumnKindTag, Vec<f64>), String>((col_schema.kind, column))
1942 })
1943 .collect::<Result<Vec<(ColumnKindTag, Vec<f64>)>, String>>()?;
1944
1945 let mut column_kinds = Vec::<ColumnKindTag>::with_capacity(p);
1946 let mut values = Array2::<f64>::zeros((n, p));
1947 for (j, (kind, column)) in encoded_columns.into_iter().enumerate() {
1948 column_kinds.push(kind);
1949 values
1950 .column_mut(j)
1951 .assign(&ndarray::ArrayView1::from(&column));
1952 }
1953
1954 Ok(EncodedDataset {
1955 headers,
1956 values,
1957 schema: schema.clone(),
1958 column_kinds,
1959 })
1960}
1961
1962fn encode_one_column(
1969 name: &str,
1970 records: &[StringRecord],
1971 j: usize,
1972 col_schema: &SchemaColumn,
1973 unseen_policy: &UnseenCategoryPolicy,
1974) -> Result<Vec<f64>, String> {
1975 let level_map = if matches!(col_schema.kind, ColumnKindTag::Categorical) {
1976 Some(
1977 col_schema
1978 .levels
1979 .iter()
1980 .enumerate()
1981 .map(|(idx, v)| (v.as_str(), idx as f64))
1982 .collect::<HashMap<_, _>>(),
1983 )
1984 } else {
1985 None
1986 };
1987
1988 let mut column = Vec::<f64>::with_capacity(records.len());
1989 for (i, rec) in records.iter().enumerate() {
1990 let raw = rec
1991 .get(j)
1992 .ok_or_else(|| {
1993 String::from(DataError::SchemaMismatch {
1994 reason: format!("missing field at row {}, col {}", i + 1, j + 1),
1995 })
1996 })?
1997 .trim();
1998 if raw.is_empty() {
1999 return Err(DataError::EmptyInput {
2000 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2001 }
2002 .into());
2003 }
2004 let val = match col_schema.kind {
2005 ColumnKindTag::Continuous => raw.parse::<f64>().map_err(|err| {
2006 String::from(DataError::SchemaMismatch {
2007 reason: format!(
2008 "column '{}' is continuous in schema but row {} has non-numeric value '{}': {}",
2009 name,
2010 i + 1,
2011 raw,
2012 err
2013 ),
2014 })
2015 })?,
2016 ColumnKindTag::Binary => {
2017 let v = raw.parse::<f64>().map_err(|err| {
2018 String::from(DataError::SchemaMismatch {
2019 reason: format!(
2020 "column '{}' is binary in schema but row {} has non-numeric value '{}': {}",
2021 name,
2022 i + 1,
2023 raw,
2024 err
2025 ),
2026 })
2027 })?;
2028 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2029 return Err(DataError::SchemaMismatch {
2030 reason: format!(
2031 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2032 name,
2033 i + 1,
2034 v
2035 ),
2036 }
2037 .into());
2038 }
2039 v
2040 }
2041 ColumnKindTag::Categorical => {
2042 let map = level_map.as_ref().ok_or_else(|| {
2043 String::from(DataError::EncodingFailure {
2044 reason: "internal categorical schema map missing".to_string(),
2045 })
2046 })?;
2047 match map.get(raw) {
2048 Some(v) => *v,
2049 None => unseen_policy
2050 .unseen_code_for(name, col_schema.levels.len())
2051 .ok_or_else(|| {
2052 String::from(DataError::SchemaMismatch {
2053 reason: format!(
2054 "unseen level '{}' in categorical column '{}' at row {}; allowed levels: {}",
2055 raw,
2056 name,
2057 i + 1,
2058 col_schema.levels.join(",")
2059 ),
2060 })
2061 })?,
2062 }
2063 }
2064 };
2065 if !val.is_finite() {
2066 return Err(DataError::InvalidValue {
2067 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2068 }
2069 .into());
2070 }
2071 column.push(val);
2072 }
2073 Ok(column)
2074}
2075
2076fn infer_schema_column(
2077 name: &str,
2078 records: &[StringRecord],
2079 col_idx: usize,
2080) -> Result<SchemaColumn, DataError> {
2081 let mut all_numeric = true;
2082 let mut all_binary = true;
2083 let mut levels = Vec::<String>::new();
2084 let mut level_index = HashMap::<String, usize>::new();
2085 for (i, rec) in records.iter().enumerate() {
2086 let raw = rec
2087 .get(col_idx)
2088 .ok_or_else(|| DataError::SchemaMismatch {
2089 reason: format!("missing field at row {}, col {}", i + 1, col_idx + 1),
2090 })?
2091 .trim();
2092 if raw.is_empty() {
2093 return Err(DataError::EmptyInput {
2094 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2095 });
2096 }
2097 if let Ok(v) = raw.parse::<f64>() {
2098 if !v.is_finite() {
2099 return Err(DataError::InvalidValue {
2100 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2101 });
2102 }
2103 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2104 all_binary = false;
2105 }
2106 } else {
2107 all_numeric = false;
2108 all_binary = false;
2109 level_index.entry(raw.to_string()).or_insert_with(|| {
2110 let idx = levels.len();
2111 levels.push(raw.to_string());
2112 idx
2113 });
2114 }
2115 }
2116 let kind = if all_numeric {
2117 if all_binary {
2118 ColumnKindTag::Binary
2119 } else {
2120 ColumnKindTag::Continuous
2121 }
2122 } else {
2123 ColumnKindTag::Categorical
2124 };
2125 if matches!(kind, ColumnKindTag::Categorical) {
2130 sort_levels_canonical(&mut levels);
2131 }
2132 Ok(SchemaColumn {
2133 name: name.to_string(),
2134 kind,
2135 levels: if matches!(kind, ColumnKindTag::Categorical) {
2136 levels
2137 } else {
2138 Vec::new()
2139 },
2140 })
2141}
2142
2143pub fn infer_and_encode_column_major(
2156 name: &str,
2157 column: &[&str],
2158 col_index: usize,
2159) -> Result<(SchemaColumn, Vec<f64>), String> {
2160 if column.is_empty() {
2161 return Err(DataError::EmptyInput {
2162 reason: "table data cannot be empty".to_string(),
2163 }
2164 .into());
2165 }
2166 let force_categorical = column.iter().any(|c| strip_categorical_sentinel(c).1);
2171 let mut all_numeric = !force_categorical;
2172 let mut all_binary = !force_categorical;
2173 let mut levels = Vec::<String>::new();
2174 let mut level_index = HashMap::<String, usize>::new();
2175 let mut trimmed = Vec::<&str>::with_capacity(column.len());
2176 let mut parsed = Vec::<Option<f64>>::with_capacity(column.len());
2183 for (i, raw_field) in column.iter().enumerate() {
2184 let (raw, _) = strip_categorical_sentinel(raw_field);
2187 let raw = raw.trim();
2188 if raw.is_empty() {
2189 return Err(DataError::EmptyInput {
2190 reason: format!("empty field at row {}, column '{}'", i + 1, name),
2191 }
2192 .into());
2193 }
2194 if !force_categorical {
2197 if let Ok(v) = raw.parse::<f64>() {
2198 if !v.is_finite() {
2199 return Err(DataError::InvalidValue {
2200 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2201 }
2202 .into());
2203 }
2204 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2205 all_binary = false;
2206 }
2207 parsed.push(Some(v));
2208 trimmed.push(raw);
2209 continue;
2210 }
2211 all_numeric = false;
2212 all_binary = false;
2213 }
2214 level_index.entry(raw.to_string()).or_insert_with(|| {
2215 let idx = levels.len();
2216 levels.push(raw.to_string());
2217 idx
2218 });
2219 parsed.push(None);
2220 trimmed.push(raw);
2221 }
2222 let kind = if all_numeric {
2223 if all_binary {
2224 ColumnKindTag::Binary
2225 } else {
2226 ColumnKindTag::Continuous
2227 }
2228 } else {
2229 ColumnKindTag::Categorical
2230 };
2231 if matches!(kind, ColumnKindTag::Categorical) {
2245 sort_levels_canonical(&mut levels);
2246 }
2247 let schema = SchemaColumn {
2248 name: name.to_string(),
2249 kind,
2250 levels: if matches!(kind, ColumnKindTag::Categorical) {
2251 levels
2252 } else {
2253 Vec::new()
2254 },
2255 };
2256
2257 let level_map = if matches!(kind, ColumnKindTag::Categorical) {
2258 Some(
2259 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 values = Vec::<f64>::with_capacity(trimmed.len());
2271 for (i, raw) in trimmed.iter().enumerate() {
2272 let raw = *raw;
2273 let val = match kind {
2274 ColumnKindTag::Continuous => parsed[i].ok_or_else(|| {
2278 String::from(DataError::EncodingFailure {
2279 reason: format!(
2280 "internal: continuous column '{}' lost its parsed value at row {} (col {})",
2281 name,
2282 i + 1,
2283 col_index
2284 ),
2285 })
2286 })?,
2287 ColumnKindTag::Binary => {
2288 let v = parsed[i].ok_or_else(|| {
2289 String::from(DataError::EncodingFailure {
2290 reason: format!(
2291 "internal: binary column '{}' lost its parsed value at row {} (col {})",
2292 name,
2293 i + 1,
2294 col_index
2295 ),
2296 })
2297 })?;
2298 if (v - 0.0).abs() >= 1e-12 && (v - 1.0).abs() >= 1e-12 {
2299 return Err(DataError::SchemaMismatch {
2300 reason: format!(
2301 "column '{}' is binary in schema but row {} has value {}; expected 0 or 1",
2302 name,
2303 i + 1,
2304 v
2305 ),
2306 }
2307 .into());
2308 }
2309 v
2310 }
2311 ColumnKindTag::Categorical => {
2312 let map = level_map.as_ref().ok_or_else(|| {
2313 String::from(DataError::EncodingFailure {
2314 reason: "internal categorical schema map missing".to_string(),
2315 })
2316 })?;
2317 *map.get(raw).ok_or_else(|| {
2318 String::from(DataError::EncodingFailure {
2319 reason: format!(
2320 "internal: level '{}' missing from freshly built map for column '{}' (col {})",
2321 raw, name, col_index
2322 ),
2323 })
2324 })?
2325 }
2326 };
2327 if !val.is_finite() {
2328 return Err(DataError::InvalidValue {
2329 reason: format!("non-finite value at row {}, column '{}'", i + 1, name),
2330 }
2331 .into());
2332 }
2333 values.push(val);
2334 }
2335 Ok((schema, values))
2336}
2337
2338#[cfg(test)]
2339mod tests {
2340 use super::*;
2341
2342 #[test]
2343 fn encode_records_rejects_empty_input() {
2344 let headers = vec!["x".to_string()];
2345 let schema = DataSchema {
2346 columns: vec![SchemaColumn {
2347 name: "x".to_string(),
2348 kind: ColumnKindTag::Continuous,
2349 levels: Vec::new(),
2350 }],
2351 };
2352
2353 let err = encode_recordswith_inferred_schema(headers.clone(), Vec::new())
2354 .expect_err("empty inferred records should error");
2355 assert_eq!(err, "table data cannot be empty");
2356
2357 let err =
2358 encode_recordswith_schema(headers, Vec::new(), &schema, UnseenCategoryPolicy::Error)
2359 .expect_err("empty schema-guided records should error");
2360 assert_eq!(err, "table data cannot be empty");
2361 }
2362
2363 #[test]
2364 fn column_major_matches_record_driven_inferred_encode() {
2365 let headers = vec!["cont".to_string(), "bin".to_string(), "cat".to_string()];
2370 let raw_rows = vec![
2371 vec!["1.5", "0", "a"],
2372 vec!["2.0", "1", "b"],
2373 vec!["-3.25", "1", "a"],
2374 vec!["0.0", "0", "c"],
2375 ];
2376 let records: Vec<StringRecord> = raw_rows
2377 .iter()
2378 .map(|r| StringRecord::from(r.clone()))
2379 .collect();
2380 let record_ds = encode_recordswith_inferred_schema(headers.clone(), records)
2381 .expect("record-driven encode");
2382
2383 for (j, name) in headers.iter().enumerate() {
2384 let column: Vec<&str> = raw_rows.iter().map(|r| r[j]).collect();
2385 let (schema_col, values) =
2386 infer_and_encode_column_major(name, &column, j + 1).expect("column-major encode");
2387 assert_eq!(schema_col.kind, record_ds.schema.columns[j].kind);
2388 assert_eq!(schema_col.levels, record_ds.schema.columns[j].levels);
2389 for (i, v) in values.iter().enumerate() {
2390 assert_eq!(*v, record_ds.values[[i, j]], "row {i} col {name}");
2391 }
2392 }
2393 }
2394
2395 #[test]
2396 fn encode_records_can_encode_unseen_named_categorical_column() {
2397 let schema = DataSchema {
2398 columns: vec![
2399 SchemaColumn {
2400 name: "g".to_string(),
2401 kind: ColumnKindTag::Categorical,
2402 levels: vec!["a".to_string(), "b".to_string()],
2403 },
2404 SchemaColumn {
2405 name: "x".to_string(),
2406 kind: ColumnKindTag::Categorical,
2407 levels: vec!["low".to_string(), "high".to_string()],
2408 },
2409 ],
2410 };
2411 let headers = vec!["g".to_string(), "x".to_string()];
2412 let records = vec![StringRecord::from(vec!["new-group", "low"])];
2413 let policy =
2414 UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
2415
2416 let ds =
2417 encode_recordswith_schema(headers, records, &schema, policy).expect("encoded dataset");
2418
2419 assert_eq!(ds.values[[0, 0]], 2.0);
2420 assert_eq!(ds.values[[0, 1]], 0.0);
2421 }
2422
2423 #[test]
2424 fn numeric_valued_dictionary_column_classifies_and_decodes_as_numeric() {
2425 use arrow::array::{Array, ArrayRef, DictionaryArray, Int8Array, Int64Array};
2435 use arrow::datatypes::{DataType, Int8Type};
2436 use std::sync::Arc;
2437
2438 let keys = Int8Array::from(vec![0i8, 1, 0, 1, 0]);
2440 let dict_values: ArrayRef = Arc::new(Int64Array::from(vec![5i64, 7]));
2441 let dict: DictionaryArray<Int8Type> = DictionaryArray::new(keys, dict_values);
2442
2443 assert!(matches!(dict.data_type(), DataType::Dictionary(_, _)));
2446 assert!(
2447 !parquet_field_is_string(dict.data_type()),
2448 "Dictionary(Int8, Int64) must not be treated as a string column"
2449 );
2450
2451 let str_dict: DictionaryArray<Int8Type> = vec!["a", "b", "a"].into_iter().collect();
2453 assert!(
2454 parquet_field_is_string(str_dict.data_type()),
2455 "Dictionary(Int8, Utf8) must remain a string column"
2456 );
2457
2458 let decoded = decode_parquet_batch_column(&dict, dict.len(), 0, "x", false)
2462 .expect("numeric dictionary column should decode as numeric");
2463 match decoded {
2464 ParquetBatchColumn::Numeric(values) => {
2465 assert_eq!(values, vec![5.0, 7.0, 5.0, 7.0, 5.0]);
2466 }
2467 ParquetBatchColumn::Strings(_) => {
2468 panic!("numeric dictionary column was decoded as strings");
2469 }
2470 }
2471
2472 use arrow::datatypes::{Field, Schema};
2477 use arrow::record_batch::RecordBatch;
2478 use parquet::arrow::ArrowWriter;
2479
2480 let arrow_schema = Arc::new(Schema::new(vec![Field::new(
2481 "x",
2482 dict.data_type().clone(),
2483 false,
2484 )]));
2485 let batch = RecordBatch::try_new(arrow_schema.clone(), vec![Arc::new(dict.clone())])
2486 .expect("record batch with a dictionary numeric column");
2487
2488 let dir = tempfile::tempdir().expect("tempdir");
2489 let path = dir.path().join("dict_numeric.parquet");
2490 {
2491 let file = std::fs::File::create(&path).expect("create parquet");
2492 let mut writer =
2493 ArrowWriter::try_new(file, arrow_schema, None).expect("arrow parquet writer");
2494 writer.write(&batch).expect("write batch");
2495 writer.close().expect("close writer");
2496 }
2497
2498 let inferred =
2501 load_parquet_inferred(&path, &[], &HashSet::new()).expect("inferred parquet load");
2502 assert_eq!(inferred.column_kinds, vec![ColumnKindTag::Continuous]);
2503 assert_eq!(
2504 inferred.values.column(0).to_vec(),
2505 vec![5.0, 7.0, 5.0, 7.0, 5.0]
2506 );
2507
2508 let schema = DataSchema {
2512 columns: vec![SchemaColumn {
2513 name: "x".to_string(),
2514 kind: ColumnKindTag::Continuous,
2515 levels: Vec::new(),
2516 }],
2517 };
2518 let schema_loaded =
2519 load_parquet_with_schema(&path, &schema, UnseenCategoryPolicy::Error, &[])
2520 .expect("dictionary-encoded numeric parquet must load against a Continuous schema");
2521 assert_eq!(schema_loaded.column_kinds, vec![ColumnKindTag::Continuous]);
2522 assert_eq!(
2523 schema_loaded.values.column(0).to_vec(),
2524 vec![5.0, 7.0, 5.0, 7.0, 5.0]
2525 );
2526 }
2527
2528 #[test]
2529 fn encode_records_keeps_unlisted_categorical_columns_strict() {
2530 let schema = DataSchema {
2531 columns: vec![
2532 SchemaColumn {
2533 name: "g".to_string(),
2534 kind: ColumnKindTag::Categorical,
2535 levels: vec!["a".to_string(), "b".to_string()],
2536 },
2537 SchemaColumn {
2538 name: "x".to_string(),
2539 kind: ColumnKindTag::Categorical,
2540 levels: vec!["low".to_string(), "high".to_string()],
2541 },
2542 ],
2543 };
2544 let headers = vec!["g".to_string(), "x".to_string()];
2545 let records = vec![StringRecord::from(vec!["a", "new-level"])];
2546 let policy =
2547 UnseenCategoryPolicy::encode_unknown_for_columns(HashSet::from(["g".to_string()]));
2548
2549 let err = encode_recordswith_schema(headers, records, &schema, policy)
2550 .expect_err("ordinary categorical column should stay strict");
2551
2552 assert!(err.contains("unseen level 'new-level' in categorical column 'x'"));
2553 }
2554
2555 #[test]
2560 fn sentinel_strip_present_returns_rest_and_true() {
2561 let marked = format!("{}{}", CATEGORICAL_CELL_SENTINEL, "hello");
2562 let (rest, found) = strip_categorical_sentinel(&marked);
2563 assert_eq!(rest, "hello");
2564 assert!(found);
2565 }
2566
2567 #[test]
2568 fn sentinel_strip_absent_returns_original_and_false() {
2569 let (rest, found) = strip_categorical_sentinel("hello");
2570 assert_eq!(rest, "hello");
2571 assert!(!found);
2572 }
2573
2574 #[test]
2575 fn sentinel_strip_empty_string_returns_empty_and_false() {
2576 let (rest, found) = strip_categorical_sentinel("");
2577 assert_eq!(rest, "");
2578 assert!(!found);
2579 }
2580
2581 #[test]
2582 fn sentinel_strip_only_sentinel_returns_empty_and_true() {
2583 let marked = CATEGORICAL_CELL_SENTINEL.to_string();
2584 let (rest, found) = strip_categorical_sentinel(&marked);
2585 assert_eq!(rest, "");
2586 assert!(found);
2587 }
2588
2589 #[test]
2594 fn feature_ranges_two_columns() {
2595 let values = ndarray::arr2(&[[1.0_f64, 10.0], [3.0, 20.0], [2.0, 15.0]]);
2596 let ds = EncodedDataset {
2597 headers: vec!["a".to_string(), "b".to_string()],
2598 values,
2599 schema: DataSchema { columns: vec![] },
2600 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
2601 };
2602 let ranges = ds.feature_ranges();
2603 assert_eq!(ranges.len(), 2);
2604 assert_eq!(ranges[0], (1.0, 3.0));
2605 assert_eq!(ranges[1], (10.0, 20.0));
2606 }
2607
2608 #[test]
2609 fn feature_ranges_single_row_min_equals_max() {
2610 let values = ndarray::arr2(&[[5.0_f64, -3.0]]);
2611 let ds = EncodedDataset {
2612 headers: vec!["x".to_string(), "y".to_string()],
2613 values,
2614 schema: DataSchema { columns: vec![] },
2615 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
2616 };
2617 let ranges = ds.feature_ranges();
2618 assert_eq!(ranges[0], (5.0, 5.0));
2619 assert_eq!(ranges[1], (-3.0, -3.0));
2620 }
2621
2622 #[test]
2623 fn feature_ranges_all_nan_defaults_to_zero() {
2624 let values = ndarray::arr2(&[[f64::NAN], [f64::NAN]]);
2625 let ds = EncodedDataset {
2626 headers: vec!["x".to_string()],
2627 values,
2628 schema: DataSchema { columns: vec![] },
2629 column_kinds: vec![ColumnKindTag::Continuous],
2630 };
2631 let ranges = ds.feature_ranges();
2632 assert_eq!(ranges[0], (0.0, 0.0));
2633 }
2634
2635 #[test]
2640 fn column_map_indexes_by_name() {
2641 let values = ndarray::arr2(&[[0.0_f64, 1.0], [2.0, 3.0]]);
2642 let ds = EncodedDataset {
2643 headers: vec!["alpha".to_string(), "beta".to_string()],
2644 values,
2645 schema: DataSchema { columns: vec![] },
2646 column_kinds: vec![ColumnKindTag::Continuous, ColumnKindTag::Continuous],
2647 };
2648 let map = ds.column_map();
2649 assert_eq!(map["alpha"], 0);
2650 assert_eq!(map["beta"], 1);
2651 assert_eq!(map.len(), 2);
2652 }
2653
2654 #[test]
2657 fn shared_prefix_identical_strings() {
2658 assert_eq!(shared_prefix("hello", "hello"), 5);
2659 }
2660
2661 #[test]
2662 fn shared_prefix_no_common_prefix() {
2663 assert_eq!(shared_prefix("abc", "xyz"), 0);
2664 }
2665
2666 #[test]
2667 fn shared_prefix_partial_match() {
2668 assert_eq!(shared_prefix("foobar", "foobaz"), 5);
2669 }
2670
2671 #[test]
2672 fn shared_prefix_one_empty() {
2673 assert_eq!(shared_prefix("", "hello"), 0);
2674 assert_eq!(shared_prefix("hello", ""), 0);
2675 }
2676
2677 #[test]
2678 fn shared_prefix_both_empty() {
2679 assert_eq!(shared_prefix("", ""), 0);
2680 }
2681
2682 #[test]
2683 fn shared_prefix_shorter_string_is_prefix() {
2684 assert_eq!(shared_prefix("foo", "foobar"), 3);
2685 }
2686
2687 #[test]
2690 fn detect_format_csv() {
2691 let path = std::path::Path::new("data.csv");
2692 assert_eq!(detect_format(path).unwrap(), DataFormat::Csv);
2693 }
2694
2695 #[test]
2696 fn detect_format_tsv() {
2697 assert_eq!(
2698 detect_format(std::path::Path::new("data.tsv")).unwrap(),
2699 DataFormat::Tsv
2700 );
2701 assert_eq!(
2702 detect_format(std::path::Path::new("data.txt")).unwrap(),
2703 DataFormat::Tsv
2704 );
2705 assert_eq!(
2706 detect_format(std::path::Path::new("data.tab")).unwrap(),
2707 DataFormat::Tsv
2708 );
2709 }
2710
2711 #[test]
2712 fn detect_format_parquet() {
2713 assert_eq!(
2714 detect_format(std::path::Path::new("data.parquet")).unwrap(),
2715 DataFormat::Parquet
2716 );
2717 assert_eq!(
2718 detect_format(std::path::Path::new("data.pq")).unwrap(),
2719 DataFormat::Parquet
2720 );
2721 assert_eq!(
2722 detect_format(std::path::Path::new("data.pqt")).unwrap(),
2723 DataFormat::Parquet
2724 );
2725 }
2726
2727 #[test]
2728 fn detect_format_uppercase_extension() {
2729 assert_eq!(
2730 detect_format(std::path::Path::new("data.CSV")).unwrap(),
2731 DataFormat::Csv
2732 );
2733 }
2734
2735 #[test]
2736 fn detect_format_unknown_extension_is_error() {
2737 let err = detect_format(std::path::Path::new("data.json")).unwrap_err();
2738 let msg = format!("{err:?}");
2739 assert!(
2740 msg.contains("json") || msg.contains("unsupported"),
2741 "error should mention extension, got: {msg}"
2742 );
2743 }
2744
2745 #[test]
2748 fn strip_categorical_sentinel_marked_cell() {
2749 let marked = "\u{0}hello";
2751 let (text, found) = strip_categorical_sentinel(marked);
2752 assert!(found);
2753 assert_eq!(text, "hello");
2754 }
2755
2756 #[test]
2757 fn strip_categorical_sentinel_unmarked_cell() {
2758 let (text, found) = strip_categorical_sentinel("plain");
2759 assert!(!found);
2760 assert_eq!(text, "plain");
2761 }
2762
2763 #[test]
2764 fn strip_categorical_sentinel_empty_string() {
2765 let (text, found) = strip_categorical_sentinel("");
2766 assert!(!found);
2767 assert_eq!(text, "");
2768 }
2769
2770 #[test]
2771 fn strip_categorical_sentinel_only_sentinel() {
2772 let s = "\u{0}";
2773 let (text, found) = strip_categorical_sentinel(s);
2774 assert!(found);
2775 assert_eq!(text, "");
2776 }
2777
2778 #[test]
2781 fn projected_headers_selects_by_index() {
2782 let all = vec![
2783 "a".to_string(),
2784 "b".to_string(),
2785 "c".to_string(),
2786 "d".to_string(),
2787 ];
2788 let selected = projected_headers(&all, &[1, 3]);
2789 assert_eq!(selected, vec!["b".to_string(), "d".to_string()]);
2790 }
2791
2792 #[test]
2793 fn projected_headers_empty_selection() {
2794 let all = vec!["x".to_string(), "y".to_string()];
2795 let selected = projected_headers(&all, &[]);
2796 assert!(selected.is_empty());
2797 }
2798
2799 #[test]
2800 fn projected_headers_all_indices() {
2801 let all = vec!["p".to_string(), "q".to_string()];
2802 let selected = projected_headers(&all, &[0, 1]);
2803 assert_eq!(selected, all);
2804 }
2805
2806 #[test]
2807 fn canonical_level_bits_collapses_signed_zero() {
2808 let pos = 0.0_f64;
2812 let neg = -0.0_f64;
2813 assert_ne!(pos.to_bits(), neg.to_bits(), "precondition: raw bits differ");
2814 assert_eq!(pos, neg, "precondition: numerically equal");
2815 assert_eq!(canonical_level_bits(pos), canonical_level_bits(neg));
2816 assert_eq!(canonical_level_bits(neg), 0.0_f64.to_bits());
2817 assert_eq!(canonical_level_bits(-1.0 * 0.0), 0.0_f64.to_bits());
2819 assert_eq!(canonical_level_bits(0.0 - 0.0), 0.0_f64.to_bits());
2820 }
2821
2822 #[test]
2823 fn canonical_level_bits_is_bit_stable_on_ordinary_values() {
2824 for &v in &[1.0_f64, -1.0, 2.5, -3.75, 1e300, -1e-300, f64::MIN, f64::MAX] {
2827 assert_eq!(canonical_level_bits(v), v.to_bits(), "value {v}");
2828 }
2829 assert_ne!(canonical_level_bits(1.0), canonical_level_bits(2.0));
2831 assert_ne!(canonical_level_bits(0.0), canonical_level_bits(1.0));
2832 assert_ne!(
2834 canonical_level_bits(f64::INFINITY),
2835 canonical_level_bits(f64::NEG_INFINITY)
2836 );
2837 }
2838
2839 #[test]
2840 fn canonical_level_bits_collapses_nan_payloads() {
2841 let a = f64::NAN;
2843 let b = f64::from_bits(0x7ff8_0000_0000_0001); let c = -f64::NAN; assert!(a.is_nan() && b.is_nan() && c.is_nan());
2846 assert_eq!(canonical_level_bits(a), canonical_level_bits(b));
2847 assert_eq!(canonical_level_bits(a), canonical_level_bits(c));
2848 }
2849
2850 #[test]
2851 fn canonical_level_bits_is_idempotent() {
2852 for &v in &[0.0_f64, -0.0, 1.0, -2.0, f64::NAN] {
2855 let once = canonical_level_bits(v);
2856 let twice = canonical_level_bits(f64::from_bits(once));
2857 assert_eq!(once, twice, "value {v}");
2858 }
2859 }
2860}