1use std::collections::{BTreeMap, BTreeSet};
2
3use serde::{Deserialize, Serialize};
4
5use crate::content_hash::StreamingHasher;
6use crate::coordinator::CoordinatorRelationSet;
7use crate::error::{DataError, Result};
8use crate::handle::{CoordinatorFeatureBlock, CoordinatorFeatureBlockF64, CoordinatorFeatureTable};
9use crate::ids::{ObservationId, RepresentationId, SourceId};
10
11pub const NUMERIC_FEATURE_BUFFER_MANIFEST_SCHEMA_VERSION: u32 = 1;
12
13#[derive(Clone, Debug, PartialEq)]
14pub struct NumericFeatureBuffer {
15 pub feature_set_id: String,
16 pub representation_id: RepresentationId,
17 pub feature_names: Vec<String>,
18 pub observation_ids: Vec<ObservationId>,
19 columns: Vec<Vec<Option<f64>>>,
20 row_index_by_observation: BTreeMap<ObservationId, usize>,
21}
22
23#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
24pub struct NumericFeatureBufferManifest {
25 pub schema_version: u32,
26 pub feature_set_id: String,
27 pub representation_id: RepresentationId,
28 pub feature_names: Vec<String>,
29 pub observation_ids: Vec<ObservationId>,
30 pub row_count: usize,
31 pub feature_count: usize,
32 pub value_count: usize,
33 pub estimated_value_bytes: usize,
34 pub buffer_fingerprint: String,
35}
36
37#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
38pub struct NumericFeatureBufferBinding {
39 pub feature_set_id: String,
40 pub representation_id: RepresentationId,
41 pub source_ids: Vec<SourceId>,
42 pub row_count: usize,
43 pub feature_count: usize,
44 pub buffer_fingerprint: String,
45}
46
47#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
48pub struct NumericFeatureMatrixF64 {
49 pub feature_set_id: String,
50 pub representation_id: RepresentationId,
51 pub feature_names: Vec<String>,
52 pub observation_ids: Vec<ObservationId>,
53 pub values: Vec<f64>,
54 #[serde(default)]
55 pub validity_mask: Option<Vec<bool>>,
56}
57
58#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
59pub struct NumericFeatureMatrixF64Columnar {
60 pub feature_set_id: String,
61 pub representation_id: RepresentationId,
62 pub feature_names: Vec<String>,
63 pub observation_ids: Vec<ObservationId>,
64 pub columns: Vec<Vec<f64>>,
65 #[serde(default)]
66 pub validity_masks: Option<Vec<Vec<bool>>>,
67}
68
69#[derive(Clone, Debug, Default, PartialEq)]
70pub struct NumericFeatureBufferStore {
71 buffers: BTreeMap<String, NumericFeatureBuffer>,
72}
73
74#[derive(Clone, Debug, Default, PartialEq)]
75pub struct NumericFeatureBufferArena {
76 store: NumericFeatureBufferStore,
77 data_bindings: BTreeMap<u64, BTreeMap<String, NumericFeatureBufferBinding>>,
78}
79
80impl NumericFeatureBuffer {
81 pub fn validate(&self) -> Result<()> {
84 if self.feature_set_id.trim().is_empty()
85 || self.feature_names.is_empty()
86 || self.feature_names.len() != self.columns.len()
87 || self.observation_ids.is_empty()
88 || self.observation_ids.len() != self.row_index_by_observation.len()
89 {
90 return Err(DataError::Validation(
91 "feature buffer descriptors do not match storage".into(),
92 ));
93 }
94 let mut names = BTreeSet::new();
95 if self
96 .feature_names
97 .iter()
98 .any(|name| name.trim().is_empty() || !names.insert(name))
99 {
100 return Err(DataError::Validation(
101 "feature buffer contains empty or duplicate feature names".into(),
102 ));
103 }
104 for (idx, observation) in self.observation_ids.iter().enumerate() {
105 if self.row_index_by_observation.get(observation) != Some(&idx) {
106 return Err(DataError::Validation(
107 "feature buffer observation descriptors do not match row index".into(),
108 ));
109 }
110 }
111 if self
112 .columns
113 .iter()
114 .any(|column| column.len() != self.observation_ids.len())
115 {
116 return Err(DataError::Validation(
117 "feature buffer column length differs from observation count".into(),
118 ));
119 }
120 Ok(())
121 }
122
123 pub fn from_feature_table(table: CoordinatorFeatureTable) -> Result<Self> {
124 table.validate()?;
125 let row_count = table.rows.len();
126 let mut observation_ids = Vec::with_capacity(row_count);
127 let mut columns = (0..table.feature_names.len())
128 .map(|_| Vec::with_capacity(row_count))
129 .collect::<Vec<_>>();
130 let mut row_index_by_observation = BTreeMap::new();
131
132 for (row_idx, row) in table.rows.into_iter().enumerate() {
133 if row_index_by_observation
134 .insert(row.observation_id.clone(), row_idx)
135 .is_some()
136 {
137 return Err(DataError::Validation(format!(
138 "feature table `{}` contains duplicate observation `{}`",
139 table.feature_set_id, row.observation_id
140 )));
141 }
142 observation_ids.push(row.observation_id.clone());
143 for (feature_idx, value) in row.values.into_iter().enumerate() {
144 let feature_name = &table.feature_names[feature_idx];
145 columns[feature_idx].push(numeric_feature_value(
146 &table.feature_set_id,
147 &row.observation_id,
148 feature_name,
149 value,
150 )?);
151 }
152 }
153
154 Ok(Self {
155 feature_set_id: table.feature_set_id,
156 representation_id: table.representation_id,
157 feature_names: table.feature_names,
158 observation_ids,
159 columns,
160 row_index_by_observation,
161 })
162 }
163
164 pub fn from_f64_matrix(matrix: NumericFeatureMatrixF64) -> Result<Self> {
165 matrix.validate()?;
166 let row_count = matrix.observation_ids.len();
167 let feature_count = matrix.feature_names.len();
168 let mut columns = (0..feature_count)
169 .map(|_| Vec::with_capacity(row_count))
170 .collect::<Vec<_>>();
171 for row_idx in 0..row_count {
172 for (feature_idx, column) in columns.iter_mut().enumerate() {
173 let flat_idx = row_idx * feature_count + feature_idx;
174 let is_valid = matrix
175 .validity_mask
176 .as_ref()
177 .map(|mask| mask[flat_idx])
178 .unwrap_or(true);
179 column.push(is_valid.then_some(matrix.values[flat_idx]));
180 }
181 }
182 let row_index_by_observation = matrix
183 .observation_ids
184 .iter()
185 .cloned()
186 .enumerate()
187 .map(|(idx, observation_id)| (observation_id, idx))
188 .collect();
189
190 Ok(Self {
191 feature_set_id: matrix.feature_set_id,
192 representation_id: matrix.representation_id,
193 feature_names: matrix.feature_names,
194 observation_ids: matrix.observation_ids,
195 columns,
196 row_index_by_observation,
197 })
198 }
199
200 pub fn from_f64_column_matrix(matrix: NumericFeatureMatrixF64Columnar) -> Result<Self> {
201 matrix.validate()?;
202 let row_count = matrix.observation_ids.len();
203 let mut columns = Vec::with_capacity(matrix.feature_names.len());
204 for (feature_idx, column_values) in matrix.columns.into_iter().enumerate() {
205 let mask = matrix
206 .validity_masks
207 .as_ref()
208 .map(|masks| masks[feature_idx].as_slice());
209 let mut column = Vec::with_capacity(row_count);
210 for (row_idx, value) in column_values.into_iter().enumerate() {
211 let is_valid = mask.map(|mask| mask[row_idx]).unwrap_or(true);
212 column.push(is_valid.then_some(value));
213 }
214 columns.push(column);
215 }
216 let row_index_by_observation = matrix
217 .observation_ids
218 .iter()
219 .cloned()
220 .enumerate()
221 .map(|(idx, observation_id)| (observation_id, idx))
222 .collect();
223
224 Ok(Self {
225 feature_set_id: matrix.feature_set_id,
226 representation_id: matrix.representation_id,
227 feature_names: matrix.feature_names,
228 observation_ids: matrix.observation_ids,
229 columns,
230 row_index_by_observation,
231 })
232 }
233
234 pub fn row_count(&self) -> usize {
235 self.observation_ids.len()
236 }
237
238 pub fn feature_count(&self) -> usize {
239 self.feature_names.len()
240 }
241
242 pub fn value_count(&self) -> usize {
243 self.row_count() * self.feature_count()
244 }
245
246 pub fn estimated_value_bytes(&self) -> usize {
247 self.value_count() * std::mem::size_of::<f64>()
248 }
249
250 pub fn contains_observation(&self, observation_id: &ObservationId) -> bool {
251 self.row_index_by_observation.contains_key(observation_id)
252 }
253
254 pub fn fingerprint(&self) -> Result<String> {
292 self.validate()?;
293 let mut hasher = StreamingHasher::new(b"dag-ml-data.numeric-feature-buffer.v2\0");
294 hasher.absorb_str(&self.feature_set_id);
295 hasher.absorb_str(self.representation_id.as_str());
296 hasher.absorb_str_collection(self.feature_names.iter().map(String::as_str));
297 hasher.absorb_str_collection(self.observation_ids.iter().map(ObservationId::as_str));
298 let n_rows = self.row_count();
299 let n_cols = self.feature_count();
300 hasher.absorb_len(n_rows);
301 hasher.absorb_len(n_cols);
302 for row in 0..n_rows {
303 for column in &self.columns {
304 hasher.absorb_cell(column[row]);
305 }
306 }
307 Ok(hasher.finalize_hex())
308 }
309
310 pub fn manifest(&self) -> Result<NumericFeatureBufferManifest> {
311 Ok(NumericFeatureBufferManifest {
312 schema_version: NUMERIC_FEATURE_BUFFER_MANIFEST_SCHEMA_VERSION,
313 feature_set_id: self.feature_set_id.clone(),
314 representation_id: self.representation_id.clone(),
315 feature_names: self.feature_names.clone(),
316 observation_ids: self.observation_ids.clone(),
317 row_count: self.row_count(),
318 feature_count: self.feature_count(),
319 value_count: self.value_count(),
320 estimated_value_bytes: self.estimated_value_bytes(),
321 buffer_fingerprint: self.fingerprint()?,
322 })
323 }
324
325 pub fn to_f64_column_matrix(&self) -> NumericFeatureMatrixF64Columnar {
331 let row_count = self.row_count();
332 let mut columns = Vec::with_capacity(self.columns.len());
333 let mut masks = Vec::with_capacity(self.columns.len());
334 let mut any_null = false;
335 for column in &self.columns {
336 let mut values = Vec::with_capacity(row_count);
337 let mut mask = Vec::with_capacity(row_count);
338 for cell in column {
339 match cell {
340 Some(value) => {
341 values.push(*value);
342 mask.push(true);
343 }
344 None => {
345 values.push(0.0);
346 mask.push(false);
347 any_null = true;
348 }
349 }
350 }
351 columns.push(values);
352 masks.push(mask);
353 }
354 NumericFeatureMatrixF64Columnar {
355 feature_set_id: self.feature_set_id.clone(),
356 representation_id: self.representation_id.clone(),
357 feature_names: self.feature_names.clone(),
358 observation_ids: self.observation_ids.clone(),
359 columns,
360 validity_masks: any_null.then_some(masks),
361 }
362 }
363
364 pub fn selected_indices(&self, columns: Option<&[String]>) -> Result<Vec<usize>> {
365 self.validate()?;
366 let index_by_name = self
367 .feature_names
368 .iter()
369 .enumerate()
370 .map(|(idx, name)| (name, idx))
371 .collect::<BTreeMap<_, _>>();
372 let indices = if let Some(columns) = columns {
373 let mut seen = BTreeSet::new();
374 columns
375 .iter()
376 .map(|column| {
377 if !seen.insert(column) {
378 return Err(DataError::Validation(format!(
379 "feature table `{}` selected duplicate feature column `{}`",
380 self.feature_set_id, column
381 )));
382 }
383 index_by_name.get(column).copied().ok_or_else(|| {
384 DataError::Validation(format!(
385 "feature table `{}` has no feature column `{}`",
386 self.feature_set_id, column
387 ))
388 })
389 })
390 .collect::<Result<Vec<_>>>()?
391 } else {
392 (0..self.feature_names.len()).collect()
393 };
394 if indices.is_empty() {
395 return Err(DataError::Validation(format!(
396 "feature table `{}` selected no feature columns",
397 self.feature_set_id
398 )));
399 }
400 Ok(indices)
401 }
402
403 pub fn project_relations(
404 &self,
405 relations: &CoordinatorRelationSet,
406 source_id: Option<&SourceId>,
407 columns: Option<&[String]>,
408 ) -> Result<CoordinatorFeatureBlock> {
409 relations.validate()?;
410 let selected_indices = self.selected_indices(columns)?;
411 let mut observation_ids = Vec::with_capacity(relations.records.len());
412 let mut sample_ids = Vec::with_capacity(relations.records.len());
413 let mut values = Vec::with_capacity(relations.records.len());
414
415 for relation in relations.records.iter().filter(|relation| {
416 source_id
417 .map(|source_id| relation.source_id.as_ref() == Some(source_id))
418 .unwrap_or(true)
419 }) {
420 let row_idx = self
421 .row_index_by_observation
422 .get(&relation.observation_id)
423 .ok_or_else(|| {
424 DataError::Validation(format!(
425 "feature table `{}` has no row for observation `{}`",
426 self.feature_set_id, relation.observation_id
427 ))
428 })?;
429 observation_ids.push(relation.observation_id.clone());
430 sample_ids.push(relation.sample_id.clone());
431 values.push(
432 selected_indices
433 .iter()
434 .map(|feature_idx| {
435 self.columns[*feature_idx][*row_idx]
436 .map_or(serde_json::Value::Null, serde_json::Value::from)
437 })
438 .collect(),
439 );
440 }
441
442 Ok(CoordinatorFeatureBlock {
443 feature_set_id: self.feature_set_id.clone(),
444 representation_id: self.representation_id.clone(),
445 feature_names: selected_indices
446 .iter()
447 .map(|idx| self.feature_names[*idx].clone())
448 .collect(),
449 observation_ids,
450 sample_ids,
451 values,
452 })
453 }
454
455 pub fn project_relations_f64(
461 &self,
462 relations: &CoordinatorRelationSet,
463 source_id: Option<&SourceId>,
464 columns: Option<&[String]>,
465 ) -> Result<CoordinatorFeatureBlockF64> {
466 relations.validate()?;
467 let selected_indices = self.selected_indices(columns)?;
468 let mut observation_ids = Vec::with_capacity(relations.records.len());
469 let mut sample_ids = Vec::with_capacity(relations.records.len());
470 let mut values = Vec::with_capacity(relations.records.len() * selected_indices.len());
471
472 for relation in relations.records.iter().filter(|relation| {
473 source_id
474 .map(|source_id| relation.source_id.as_ref() == Some(source_id))
475 .unwrap_or(true)
476 }) {
477 let row_idx = self
478 .row_index_by_observation
479 .get(&relation.observation_id)
480 .ok_or_else(|| {
481 DataError::Validation(format!(
482 "feature table `{}` has no row for observation `{}`",
483 self.feature_set_id, relation.observation_id
484 ))
485 })?;
486 for feature_idx in &selected_indices {
487 values.push(self.columns[*feature_idx][*row_idx].ok_or_else(|| {
488 DataError::Validation(format!(
489 "feature table `{}` observation `{}` feature `{}` is masked; the typed f64 projection requires fully numeric values",
490 self.feature_set_id,
491 relation.observation_id,
492 self.feature_names[*feature_idx]
493 ))
494 })?);
495 }
496 observation_ids.push(relation.observation_id.clone());
497 sample_ids.push(relation.sample_id.clone());
498 }
499
500 Ok(CoordinatorFeatureBlockF64 {
501 feature_set_id: self.feature_set_id.clone(),
502 representation_id: self.representation_id.clone(),
503 feature_names: selected_indices
504 .iter()
505 .map(|idx| self.feature_names[*idx].clone())
506 .collect(),
507 observation_ids,
508 sample_ids,
509 values,
510 })
511 }
512
513 fn binding_for_sources(
514 &self,
515 source_ids: Vec<SourceId>,
516 ) -> Result<NumericFeatureBufferBinding> {
517 Ok(NumericFeatureBufferBinding {
518 feature_set_id: self.feature_set_id.clone(),
519 representation_id: self.representation_id.clone(),
520 source_ids,
521 row_count: self.row_count(),
522 feature_count: self.feature_count(),
523 buffer_fingerprint: self.fingerprint()?,
524 })
525 }
526}
527
528impl NumericFeatureMatrixF64 {
529 pub fn validate(&self) -> Result<()> {
530 validate_feature_shape(
531 &self.feature_set_id,
532 &self.feature_names,
533 &self.observation_ids,
534 )?;
535 let expected_values = self.feature_names.len() * self.observation_ids.len();
536 if self.values.len() != expected_values {
537 return Err(DataError::Validation(format!(
538 "f64 feature matrix `{}` has {} values for {} observations x {} features",
539 self.feature_set_id,
540 self.values.len(),
541 self.observation_ids.len(),
542 self.feature_names.len()
543 )));
544 }
545 if let Some(validity_mask) = &self.validity_mask {
546 if validity_mask.len() != expected_values {
547 return Err(DataError::Validation(format!(
548 "f64 feature matrix `{}` validity_mask has {} values for {} observations x {} features",
549 self.feature_set_id,
550 validity_mask.len(),
551 self.observation_ids.len(),
552 self.feature_names.len()
553 )));
554 }
555 }
556 for (idx, value) in self.values.iter().enumerate() {
557 let is_valid = self
558 .validity_mask
559 .as_ref()
560 .map(|mask| mask[idx])
561 .unwrap_or(true);
562 if is_valid && !value.is_finite() {
563 return Err(DataError::Validation(format!(
564 "f64 feature matrix `{}` value {} is not finite",
565 self.feature_set_id, idx
566 )));
567 }
568 }
569 Ok(())
570 }
571}
572
573impl NumericFeatureMatrixF64Columnar {
574 pub fn validate(&self) -> Result<()> {
575 validate_feature_shape(
576 &self.feature_set_id,
577 &self.feature_names,
578 &self.observation_ids,
579 )?;
580 if self.columns.len() != self.feature_names.len() {
581 return Err(DataError::Validation(format!(
582 "f64 columnar feature matrix `{}` has {} columns for {} features",
583 self.feature_set_id,
584 self.columns.len(),
585 self.feature_names.len()
586 )));
587 }
588 if let Some(validity_masks) = &self.validity_masks {
589 if validity_masks.len() != self.feature_names.len() {
590 return Err(DataError::Validation(format!(
591 "f64 columnar feature matrix `{}` has {} validity_masks for {} features",
592 self.feature_set_id,
593 validity_masks.len(),
594 self.feature_names.len()
595 )));
596 }
597 }
598 let row_count = self.observation_ids.len();
599 for (feature_idx, column) in self.columns.iter().enumerate() {
600 if column.len() != row_count {
601 return Err(DataError::Validation(format!(
602 "f64 columnar feature matrix `{}` column {} has {} values for {} observations",
603 self.feature_set_id,
604 feature_idx,
605 column.len(),
606 row_count
607 )));
608 }
609 let mask = self
610 .validity_masks
611 .as_ref()
612 .map(|masks| masks[feature_idx].as_slice());
613 if let Some(mask) = mask {
614 if mask.len() != row_count {
615 return Err(DataError::Validation(format!(
616 "f64 columnar feature matrix `{}` column {} validity_mask has {} values for {} observations",
617 self.feature_set_id,
618 feature_idx,
619 mask.len(),
620 row_count
621 )));
622 }
623 }
624 for (row_idx, value) in column.iter().enumerate() {
625 let is_valid = mask.map(|mask| mask[row_idx]).unwrap_or(true);
626 if is_valid && !value.is_finite() {
627 return Err(DataError::Validation(format!(
628 "f64 columnar feature matrix `{}` column {} row {} is not finite",
629 self.feature_set_id, feature_idx, row_idx
630 )));
631 }
632 }
633 }
634 Ok(())
635 }
636}
637
638impl NumericFeatureBufferStore {
639 pub fn new(buffers: BTreeMap<String, NumericFeatureBuffer>) -> Result<Self> {
640 for (feature_set_id, buffer) in &buffers {
641 buffer.validate()?;
642 if feature_set_id != &buffer.feature_set_id {
643 return Err(DataError::Validation(format!(
644 "feature buffer store key `{feature_set_id}` does not match buffer feature_set_id `{}`",
645 buffer.feature_set_id
646 )));
647 }
648 }
649 Ok(Self { buffers })
650 }
651
652 pub fn from_feature_tables(tables: Vec<CoordinatorFeatureTable>) -> Result<Self> {
653 let mut buffers = BTreeMap::new();
654 for table in tables {
655 let feature_set_id = table.feature_set_id.clone();
656 let buffer = NumericFeatureBuffer::from_feature_table(table)?;
657 if buffers.insert(feature_set_id.clone(), buffer).is_some() {
658 return Err(DataError::Validation(format!(
659 "duplicate feature table `{feature_set_id}`"
660 )));
661 }
662 }
663 Self::new(buffers)
664 }
665
666 pub fn from_f64_matrices(matrices: Vec<NumericFeatureMatrixF64>) -> Result<Self> {
667 let mut buffers = BTreeMap::new();
668 for matrix in matrices {
669 let feature_set_id = matrix.feature_set_id.clone();
670 let buffer = NumericFeatureBuffer::from_f64_matrix(matrix)?;
671 if buffers.insert(feature_set_id.clone(), buffer).is_some() {
672 return Err(DataError::Validation(format!(
673 "duplicate f64 feature matrix `{feature_set_id}`"
674 )));
675 }
676 }
677 Self::new(buffers)
678 }
679
680 pub fn from_f64_column_matrices(
681 matrices: Vec<NumericFeatureMatrixF64Columnar>,
682 ) -> Result<Self> {
683 let mut buffers = BTreeMap::new();
684 for matrix in matrices {
685 let feature_set_id = matrix.feature_set_id.clone();
686 let buffer = NumericFeatureBuffer::from_f64_column_matrix(matrix)?;
687 if buffers.insert(feature_set_id.clone(), buffer).is_some() {
688 return Err(DataError::Validation(format!(
689 "duplicate f64 columnar feature matrix `{feature_set_id}`"
690 )));
691 }
692 }
693 Self::new(buffers)
694 }
695
696 pub fn is_empty(&self) -> bool {
697 self.buffers.is_empty()
698 }
699
700 pub fn len(&self) -> usize {
701 self.buffers.len()
702 }
703
704 pub fn get(&self, feature_set_id: &str) -> Option<&NumericFeatureBuffer> {
705 self.buffers.get(feature_set_id)
706 }
707
708 pub fn iter(&self) -> impl Iterator<Item = (&String, &NumericFeatureBuffer)> {
712 self.buffers.iter()
713 }
714
715 pub fn manifests(&self) -> Result<Vec<NumericFeatureBufferManifest>> {
716 self.buffers
717 .values()
718 .map(NumericFeatureBuffer::manifest)
719 .collect()
720 }
721
722 pub fn bindings_for_relations(
723 &self,
724 relations: &CoordinatorRelationSet,
725 representation_id: &RepresentationId,
726 ) -> Result<Vec<NumericFeatureBufferBinding>> {
727 relations.validate()?;
728 let source_ids = relations
729 .records
730 .iter()
731 .filter_map(|relation| relation.source_id.as_ref())
732 .collect::<BTreeSet<_>>();
733
734 let mut bindings = Vec::new();
735 for buffer in self.buffers.values() {
736 if &buffer.representation_id != representation_id {
737 continue;
738 }
739 let mut covered_sources = Vec::new();
740 if source_ids.is_empty() {
741 if relations
742 .records
743 .iter()
744 .all(|relation| buffer.contains_observation(&relation.observation_id))
745 {
746 bindings.push(buffer.binding_for_sources(Vec::new())?);
747 }
748 continue;
749 }
750 for source_id in &source_ids {
751 let source_records = relations
752 .records
753 .iter()
754 .filter(|relation| relation.source_id.as_ref() == Some(*source_id));
755 if source_records
756 .clone()
757 .all(|relation| buffer.contains_observation(&relation.observation_id))
758 {
759 covered_sources.push((*source_id).clone());
760 }
761 }
762 if !covered_sources.is_empty() {
763 bindings.push(buffer.binding_for_sources(covered_sources)?);
764 }
765 }
766 Ok(bindings)
767 }
768
769 pub fn project_relations(
770 &self,
771 feature_set_id: &str,
772 relations: &CoordinatorRelationSet,
773 source_id: Option<&SourceId>,
774 columns: Option<&[String]>,
775 ) -> Result<CoordinatorFeatureBlock> {
776 let buffer = self.buffers.get(feature_set_id).ok_or_else(|| {
777 DataError::Validation(format!("unknown feature buffer `{feature_set_id}`"))
778 })?;
779 buffer.project_relations(relations, source_id, columns)
780 }
781
782 pub fn project_relations_f64(
783 &self,
784 feature_set_id: &str,
785 relations: &CoordinatorRelationSet,
786 source_id: Option<&SourceId>,
787 columns: Option<&[String]>,
788 ) -> Result<CoordinatorFeatureBlockF64> {
789 let buffer = self.buffers.get(feature_set_id).ok_or_else(|| {
790 DataError::Validation(format!("unknown feature buffer `{feature_set_id}`"))
791 })?;
792 buffer.project_relations_f64(relations, source_id, columns)
793 }
794}
795
796impl NumericFeatureBufferArena {
797 pub fn new(store: NumericFeatureBufferStore) -> Self {
798 Self {
799 store,
800 data_bindings: BTreeMap::new(),
801 }
802 }
803
804 pub fn manifests(&self) -> Result<Vec<NumericFeatureBufferManifest>> {
805 self.store.manifests()
806 }
807
808 pub fn bind_data_handle(
809 &mut self,
810 data_handle: u64,
811 relations: &CoordinatorRelationSet,
812 representation_id: &RepresentationId,
813 ) -> Result<Vec<NumericFeatureBufferBinding>> {
814 let bindings = self
815 .store
816 .bindings_for_relations(relations, representation_id)?;
817 self.data_bindings.insert(
818 data_handle,
819 bindings
820 .iter()
821 .cloned()
822 .map(|binding| (binding.feature_set_id.clone(), binding))
823 .collect(),
824 );
825 Ok(bindings)
826 }
827
828 pub fn release_data_handle(&mut self, data_handle: u64) -> bool {
829 self.data_bindings.remove(&data_handle).is_some()
830 }
831
832 pub fn bindings_for_data_handle(
833 &self,
834 data_handle: u64,
835 ) -> Result<Vec<NumericFeatureBufferBinding>> {
836 let bindings = self.data_bindings.get(&data_handle).ok_or_else(|| {
837 DataError::Validation(format!(
838 "data handle `{data_handle}` has no feature buffer bindings"
839 ))
840 })?;
841 Ok(bindings.values().cloned().collect())
842 }
843
844 pub fn project_bound_relations(
845 &self,
846 data_handle: u64,
847 feature_set_id: &str,
848 relations: &CoordinatorRelationSet,
849 source_id: Option<&SourceId>,
850 columns: Option<&[String]>,
851 ) -> Result<CoordinatorFeatureBlock> {
852 self.validate_bound_sources(data_handle, feature_set_id, relations, source_id)?;
853 self.store
854 .project_relations(feature_set_id, relations, source_id, columns)
855 }
856
857 pub fn project_bound_relations_f64(
858 &self,
859 data_handle: u64,
860 feature_set_id: &str,
861 relations: &CoordinatorRelationSet,
862 source_id: Option<&SourceId>,
863 columns: Option<&[String]>,
864 ) -> Result<CoordinatorFeatureBlockF64> {
865 self.validate_bound_sources(data_handle, feature_set_id, relations, source_id)?;
866 self.store
867 .project_relations_f64(feature_set_id, relations, source_id, columns)
868 }
869
870 fn validate_bound_sources(
871 &self,
872 data_handle: u64,
873 feature_set_id: &str,
874 relations: &CoordinatorRelationSet,
875 source_id: Option<&SourceId>,
876 ) -> Result<()> {
877 relations.validate()?;
878 let binding = self
879 .data_bindings
880 .get(&data_handle)
881 .and_then(|bindings| bindings.get(feature_set_id))
882 .ok_or_else(|| {
883 DataError::Validation(format!(
884 "feature buffer `{feature_set_id}` is not bound to data handle `{data_handle}`"
885 ))
886 })?;
887 let relation_source_ids = relations
888 .records
889 .iter()
890 .filter_map(|relation| relation.source_id.as_ref())
891 .cloned()
892 .collect::<BTreeSet<_>>();
893 let required_source_ids = if let Some(source_id) = source_id {
894 if relation_source_ids.is_empty() || !relation_source_ids.contains(source_id) {
895 return Err(DataError::Validation(format!(
896 "feature buffer `{feature_set_id}` source `{source_id}` is not present in view for data handle `{data_handle}`"
897 )));
898 }
899 vec![source_id.clone()]
900 } else {
901 relation_source_ids.into_iter().collect::<Vec<_>>()
902 };
903 for source_id in &required_source_ids {
904 if !binding.source_ids.contains(source_id) {
905 return Err(DataError::Validation(format!(
906 "feature buffer `{feature_set_id}` is not bound to source `{source_id}` for data handle `{data_handle}`"
907 )));
908 }
909 }
910 Ok(())
911 }
912}
913
914fn numeric_feature_value(
915 feature_set_id: &str,
916 observation_id: &ObservationId,
917 feature_name: &str,
918 value: serde_json::Value,
919) -> Result<Option<f64>> {
920 match value {
921 serde_json::Value::Null => Ok(None),
922 serde_json::Value::Number(number) => number.as_f64().map(Some).ok_or_else(|| {
923 DataError::Validation(format!(
924 "feature table `{feature_set_id}` row `{observation_id}` feature `{feature_name}` contains a non-f64 numeric value"
925 ))
926 }),
927 _ => Err(DataError::Validation(format!(
928 "feature table `{feature_set_id}` row `{observation_id}` feature `{feature_name}` must be numeric or null"
929 ))),
930 }
931}
932
933fn validate_feature_shape(
934 feature_set_id: &str,
935 feature_names: &[String],
936 observation_ids: &[ObservationId],
937) -> Result<()> {
938 if feature_set_id.trim().is_empty() {
939 return Err(DataError::Validation("feature_set_id is empty".to_string()));
940 }
941 if feature_names.is_empty() {
942 return Err(DataError::Validation(format!(
943 "feature matrix `{feature_set_id}` contains no features"
944 )));
945 }
946 let mut seen_features = BTreeSet::new();
947 for feature_name in feature_names {
948 if feature_name.trim().is_empty() {
949 return Err(DataError::Validation(format!(
950 "feature matrix `{feature_set_id}` contains an empty feature name"
951 )));
952 }
953 if !seen_features.insert(feature_name) {
954 return Err(DataError::Validation(format!(
955 "feature matrix `{feature_set_id}` contains duplicate feature `{feature_name}`"
956 )));
957 }
958 }
959 if observation_ids.is_empty() {
960 return Err(DataError::Validation(format!(
961 "feature matrix `{feature_set_id}` contains no observations"
962 )));
963 }
964 let mut seen_observations = BTreeSet::new();
965 for observation_id in observation_ids {
966 if !seen_observations.insert(observation_id) {
967 return Err(DataError::Validation(format!(
968 "feature matrix `{feature_set_id}` contains duplicate observation `{observation_id}`"
969 )));
970 }
971 }
972 Ok(())
973}
974
975#[cfg(test)]
976mod tests {
977 use super::*;
978 use crate::coordinator::CoordinatorRelation;
979 use crate::handle::CoordinatorFeatureRow;
980 use crate::ids::{SampleId, TargetId};
981
982 fn oid(value: &str) -> ObservationId {
983 ObservationId::new(value).unwrap()
984 }
985
986 fn sid(value: &str) -> SampleId {
987 SampleId::new(value).unwrap()
988 }
989
990 fn source(value: &str) -> SourceId {
991 SourceId::new(value).unwrap()
992 }
993
994 fn table() -> CoordinatorFeatureTable {
995 CoordinatorFeatureTable {
996 feature_set_id: "x".to_string(),
997 representation_id: RepresentationId::new("tabular_numeric").unwrap(),
998 feature_names: vec!["f0".to_string(), "f1".to_string()],
999 rows: vec![
1000 CoordinatorFeatureRow {
1001 observation_id: oid("obs.s1.nir"),
1002 values: vec![serde_json::json!(1.0), serde_json::json!(10.0)],
1003 },
1004 CoordinatorFeatureRow {
1005 observation_id: oid("obs.s1.chem"),
1006 values: vec![serde_json::json!(2.0), serde_json::json!(20.0)],
1007 },
1008 CoordinatorFeatureRow {
1009 observation_id: oid("obs.s2.nir"),
1010 values: vec![serde_json::json!(3.0), serde_json::Value::Null],
1011 },
1012 ],
1013 }
1014 }
1015
1016 fn f64_matrix() -> NumericFeatureMatrixF64 {
1017 NumericFeatureMatrixF64 {
1018 feature_set_id: "x".to_string(),
1019 representation_id: RepresentationId::new("tabular_numeric").unwrap(),
1020 feature_names: vec!["f0".to_string(), "f1".to_string()],
1021 observation_ids: vec![oid("obs.s1.nir"), oid("obs.s1.chem"), oid("obs.s2.nir")],
1022 values: vec![1.0, 10.0, 2.0, 20.0, 3.0, 0.0],
1023 validity_mask: Some(vec![true, true, true, true, true, false]),
1024 }
1025 }
1026
1027 fn f64_column_matrix() -> NumericFeatureMatrixF64Columnar {
1028 NumericFeatureMatrixF64Columnar {
1029 feature_set_id: "x".to_string(),
1030 representation_id: RepresentationId::new("tabular_numeric").unwrap(),
1031 feature_names: vec!["f0".to_string(), "f1".to_string()],
1032 observation_ids: vec![oid("obs.s1.nir"), oid("obs.s1.chem"), oid("obs.s2.nir")],
1033 columns: vec![vec![1.0, 2.0, 3.0], vec![10.0, 20.0, 0.0]],
1034 validity_masks: Some(vec![vec![true, true, true], vec![true, true, false]]),
1035 }
1036 }
1037
1038 fn relations() -> CoordinatorRelationSet {
1039 CoordinatorRelationSet {
1040 records: vec![
1041 relation("obs.s2.nir", "S2", "nir"),
1042 relation("obs.s1.nir", "S1", "nir"),
1043 relation("obs.s1.chem", "S1", "chem"),
1044 ],
1045 }
1046 }
1047
1048 fn relation(observation_id: &str, sample_id: &str, source_id: &str) -> CoordinatorRelation {
1049 CoordinatorRelation {
1050 unit_level: crate::CoordinatorEntityUnitLevel::Observation,
1051 unit_id: None,
1052 rep_id: None,
1053 derived_unit_id: None,
1054 component_observation_ids: Vec::new(),
1055 sample_influence_weight: None,
1056 quality_flag: None,
1057 observation_id: oid(observation_id),
1058 sample_id: sid(sample_id),
1059 target_id: Some(TargetId::new("y").unwrap()),
1060 group_id: None,
1061 origin_sample_id: None,
1062 source_id: Some(source(source_id)),
1063 is_augmented: false,
1064 excluded: false,
1065 metadata: BTreeMap::new(),
1066 tags: Vec::new(),
1067 }
1068 }
1069
1070 #[test]
1071 fn projects_view_relations_from_columnar_numeric_buffer() {
1072 let buffer = NumericFeatureBuffer::from_feature_table(table()).unwrap();
1073 assert_eq!(buffer.row_count(), 3);
1074 assert_eq!(buffer.feature_count(), 2);
1075 assert_eq!(buffer.value_count(), 6);
1076 let manifest = buffer.manifest().unwrap();
1077 assert_eq!(
1078 manifest.schema_version,
1079 NUMERIC_FEATURE_BUFFER_MANIFEST_SCHEMA_VERSION
1080 );
1081 assert_eq!(manifest.row_count, 3);
1082 assert_eq!(manifest.feature_count, 2);
1083 assert_eq!(manifest.value_count, 6);
1084 assert_eq!(manifest.estimated_value_bytes, 48);
1085 assert_eq!(manifest.buffer_fingerprint.len(), 64);
1086
1087 let block = buffer
1088 .project_relations(
1089 &relations(),
1090 Some(&source("nir")),
1091 Some(&["f1".to_string()]),
1092 )
1093 .unwrap();
1094
1095 assert_eq!(block.feature_set_id, "x");
1096 assert_eq!(block.feature_names, vec!["f1".to_string()]);
1097 assert_eq!(
1098 block.observation_ids,
1099 vec![oid("obs.s2.nir"), oid("obs.s1.nir")]
1100 );
1101 assert_eq!(block.sample_ids, vec![sid("S2"), sid("S1")]);
1102 assert_eq!(
1103 block.values,
1104 vec![vec![serde_json::Value::Null], vec![serde_json::json!(10.0)]]
1105 );
1106 }
1107
1108 #[test]
1109 fn typed_f64_projection_matches_boxed_projection() {
1110 let mut matrix = f64_matrix();
1113 matrix.validity_mask = None;
1114 matrix.values = vec![1.0, 10.0, 2.0, 20.0, 3.0, 30.0];
1115 let buffer = NumericFeatureBuffer::from_f64_matrix(matrix).unwrap();
1116
1117 let boxed = buffer.project_relations(&relations(), None, None).unwrap();
1118 let typed = buffer
1119 .project_relations_f64(&relations(), None, None)
1120 .unwrap();
1121
1122 assert_eq!(typed.feature_set_id, boxed.feature_set_id);
1123 assert_eq!(typed.representation_id, boxed.representation_id);
1124 assert_eq!(typed.feature_names, boxed.feature_names);
1125 assert_eq!(typed.observation_ids, boxed.observation_ids);
1126 assert_eq!(typed.sample_ids, boxed.sample_ids);
1127 let expected: Vec<f64> = boxed
1128 .values
1129 .iter()
1130 .flat_map(|row| row.iter().map(|v| v.as_f64().unwrap()))
1131 .collect();
1132 assert_eq!(typed.values, expected);
1133 }
1134
1135 #[test]
1136 fn typed_f64_projection_rejects_masked_cells() {
1137 let buffer = NumericFeatureBuffer::from_f64_matrix(f64_matrix()).unwrap();
1140 let error = buffer
1141 .project_relations_f64(&relations(), None, None)
1142 .unwrap_err();
1143 assert!(format!("{error}").contains("is masked"));
1144 }
1145
1146 #[test]
1147 fn rejects_duplicate_selected_columns() {
1148 let buffer = NumericFeatureBuffer::from_feature_table(table()).unwrap();
1149 let error = buffer
1150 .selected_indices(Some(&["f0".to_string(), "f0".to_string()]))
1151 .unwrap_err();
1152 assert!(format!("{error}").contains("duplicate feature column"));
1153 }
1154
1155 #[test]
1156 fn rejects_missing_observation_in_projection() {
1157 let buffer = NumericFeatureBuffer::from_feature_table(table()).unwrap();
1158 let missing = CoordinatorRelationSet {
1159 records: vec![relation("obs.missing", "S9", "nir")],
1160 };
1161 let error = buffer.project_relations(&missing, None, None).unwrap_err();
1162 assert!(format!("{error}").contains("has no row for observation"));
1163 }
1164
1165 #[test]
1166 fn builds_columnar_buffer_from_row_major_f64_matrix() {
1167 let buffer = NumericFeatureBuffer::from_f64_matrix(f64_matrix()).unwrap();
1168 assert_eq!(buffer.row_count(), 3);
1169 assert_eq!(buffer.feature_count(), 2);
1170
1171 let block = buffer
1172 .project_relations(&relations(), Some(&source("nir")), None)
1173 .unwrap();
1174
1175 assert_eq!(
1176 block.observation_ids,
1177 vec![oid("obs.s2.nir"), oid("obs.s1.nir")]
1178 );
1179 assert_eq!(
1180 block.values,
1181 vec![
1182 vec![serde_json::json!(3.0), serde_json::Value::Null],
1183 vec![serde_json::json!(1.0), serde_json::json!(10.0)],
1184 ]
1185 );
1186 }
1187
1188 #[test]
1189 fn rejects_malformed_f64_matrix_shape() {
1190 let mut matrix = f64_matrix();
1191 matrix.values.pop();
1192 let error = NumericFeatureBuffer::from_f64_matrix(matrix).unwrap_err();
1193 assert!(format!("{error}").contains("has 5 values"));
1194
1195 let mut matrix = f64_matrix();
1196 matrix.validity_mask = Some(vec![true]);
1197 let error = NumericFeatureBuffer::from_f64_matrix(matrix).unwrap_err();
1198 assert!(format!("{error}").contains("validity_mask has 1 values"));
1199
1200 let mut matrix = f64_matrix();
1201 matrix.values[0] = f64::NAN;
1202 let error = NumericFeatureBuffer::from_f64_matrix(matrix).unwrap_err();
1203 assert!(format!("{error}").contains("value 0 is not finite"));
1204
1205 let mut matrix = f64_matrix();
1206 matrix.values[5] = f64::NAN;
1207 assert!(NumericFeatureBuffer::from_f64_matrix(matrix).is_ok());
1208 }
1209
1210 #[test]
1211 fn store_manifests_and_projects_by_feature_set_id() {
1212 let store = NumericFeatureBufferStore::from_feature_tables(vec![table()]).unwrap();
1213 assert_eq!(store.len(), 1);
1214 assert!(!store.is_empty());
1215
1216 let manifests = store.manifests().unwrap();
1217 assert_eq!(manifests.len(), 1);
1218 assert_eq!(manifests[0].feature_set_id, "x");
1219 assert_eq!(manifests[0].feature_names, vec!["f0", "f1"]);
1220
1221 let block = store
1222 .project_relations("x", &relations(), Some(&source("chem")), None)
1223 .unwrap();
1224 assert_eq!(block.observation_ids, vec![oid("obs.s1.chem")]);
1225 assert_eq!(
1226 block.values,
1227 vec![vec![serde_json::json!(2.0), serde_json::json!(20.0)]]
1228 );
1229 }
1230
1231 #[test]
1232 fn store_derives_source_bindings_from_relation_coverage() {
1233 let store = NumericFeatureBufferStore::from_feature_tables(vec![table()]).unwrap();
1234 let bindings = store
1235 .bindings_for_relations(
1236 &relations(),
1237 &RepresentationId::new("tabular_numeric").unwrap(),
1238 )
1239 .unwrap();
1240
1241 assert_eq!(bindings.len(), 1);
1242 assert_eq!(bindings[0].feature_set_id, "x");
1243 assert_eq!(bindings[0].source_ids, vec![source("chem"), source("nir")]);
1244 assert_eq!(bindings[0].row_count, 3);
1245 assert_eq!(bindings[0].feature_count, 2);
1246 assert_eq!(bindings[0].buffer_fingerprint.len(), 64);
1247
1248 let wrong_representation = store
1249 .bindings_for_relations(
1250 &relations(),
1251 &RepresentationId::new("dense_signal").unwrap(),
1252 )
1253 .unwrap();
1254 assert!(wrong_representation.is_empty());
1255 }
1256
1257 #[test]
1258 fn store_accepts_typed_f64_matrices() {
1259 let store = NumericFeatureBufferStore::from_f64_matrices(vec![f64_matrix()]).unwrap();
1260 let manifests = store.manifests().unwrap();
1261
1262 assert_eq!(manifests.len(), 1);
1263 assert_eq!(manifests[0].feature_set_id, "x");
1264 assert_eq!(manifests[0].value_count, 6);
1265
1266 let error = NumericFeatureBufferStore::from_f64_matrices(vec![f64_matrix(), f64_matrix()])
1267 .unwrap_err();
1268 assert!(format!("{error}").contains("duplicate f64 feature matrix"));
1269 }
1270
1271 #[test]
1272 fn arena_binds_projects_and_releases_data_handle_buffers() {
1273 let store = NumericFeatureBufferStore::from_feature_tables(vec![table()]).unwrap();
1274 let mut arena = NumericFeatureBufferArena::new(store);
1275 let bindings = arena
1276 .bind_data_handle(
1277 7,
1278 &relations(),
1279 &RepresentationId::new("tabular_numeric").unwrap(),
1280 )
1281 .unwrap();
1282
1283 assert_eq!(bindings.len(), 1);
1284 assert_eq!(arena.bindings_for_data_handle(7).unwrap(), bindings);
1285
1286 let block = arena
1287 .project_bound_relations(7, "x", &relations(), Some(&source("nir")), None)
1288 .unwrap();
1289 assert_eq!(
1290 block.observation_ids,
1291 vec![oid("obs.s2.nir"), oid("obs.s1.nir")]
1292 );
1293
1294 let error = arena
1295 .project_bound_relations(8, "x", &relations(), Some(&source("nir")), None)
1296 .unwrap_err();
1297 assert!(format!("{error}").contains("not bound to data handle"));
1298
1299 assert!(arena.release_data_handle(7));
1300 let error = arena.bindings_for_data_handle(7).unwrap_err();
1301 assert!(format!("{error}").contains("no feature buffer bindings"));
1302 }
1303
1304 #[test]
1305 fn store_refuses_duplicate_feature_sets() {
1306 let error =
1307 NumericFeatureBufferStore::from_feature_tables(vec![table(), table()]).unwrap_err();
1308 assert!(format!("{error}").contains("duplicate feature table"));
1309 }
1310
1311 #[test]
1312 fn builds_columnar_buffer_from_column_major_f64_matrix() {
1313 let buffer = NumericFeatureBuffer::from_f64_column_matrix(f64_column_matrix()).unwrap();
1314 assert_eq!(buffer.row_count(), 3);
1315 assert_eq!(buffer.feature_count(), 2);
1316
1317 let block = buffer
1318 .project_relations(&relations(), Some(&source("nir")), None)
1319 .unwrap();
1320
1321 assert_eq!(
1322 block.observation_ids,
1323 vec![oid("obs.s2.nir"), oid("obs.s1.nir")]
1324 );
1325 assert_eq!(
1326 block.values,
1327 vec![
1328 vec![serde_json::json!(3.0), serde_json::Value::Null],
1329 vec![serde_json::json!(1.0), serde_json::json!(10.0)],
1330 ]
1331 );
1332 }
1333
1334 #[test]
1335 fn column_major_and_row_major_produce_identical_buffer_fingerprints() {
1336 let row_major = NumericFeatureBuffer::from_f64_matrix(f64_matrix()).unwrap();
1337 let columnar = NumericFeatureBuffer::from_f64_column_matrix(f64_column_matrix()).unwrap();
1338 assert_eq!(
1339 row_major.fingerprint().unwrap(),
1340 columnar.fingerprint().unwrap()
1341 );
1342 }
1343
1344 fn dense_buffer(rows: usize, cols: usize) -> NumericFeatureBuffer {
1347 let columns = (0..cols)
1348 .map(|c| (0..rows).map(|r| (r * cols + c) as f64).collect::<Vec<_>>())
1349 .collect::<Vec<_>>();
1350 let matrix = NumericFeatureMatrixF64Columnar {
1351 feature_set_id: "x".to_string(),
1352 representation_id: RepresentationId::new("tabular_numeric").unwrap(),
1353 feature_names: (0..cols).map(|c| format!("f{c}")).collect(),
1354 observation_ids: (0..rows).map(|r| oid(&format!("obs.{r}"))).collect(),
1355 columns,
1356 validity_masks: None,
1357 };
1358 NumericFeatureBuffer::from_f64_column_matrix(matrix).unwrap()
1359 }
1360
1361 #[test]
1362 fn fingerprint_is_64_lowercase_hex() {
1363 let fp = NumericFeatureBuffer::from_feature_table(table())
1364 .unwrap()
1365 .fingerprint()
1366 .unwrap();
1367 assert_eq!(fp.len(), 64);
1368 assert!(fp
1369 .chars()
1370 .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()));
1371 }
1372
1373 #[test]
1374 fn fingerprint_is_deterministic_across_calls_and_clone() {
1375 let buffer = NumericFeatureBuffer::from_feature_table(table()).unwrap();
1376 let once = buffer.fingerprint().unwrap();
1377 let twice = buffer.fingerprint().unwrap();
1378 assert_eq!(once, twice, "same buffer hashed twice must match");
1379 let clone = buffer.clone();
1380 assert_eq!(
1381 once,
1382 clone.fingerprint().unwrap(),
1383 "a clone must fingerprint identically"
1384 );
1385 }
1386
1387 #[test]
1388 fn fingerprint_changes_when_a_single_cell_flips() {
1389 let baseline = f64_column_matrix();
1390 let base_fp = NumericFeatureBuffer::from_f64_column_matrix(baseline.clone())
1391 .unwrap()
1392 .fingerprint()
1393 .unwrap();
1394 let mut flipped = baseline;
1395 flipped.columns[0][0] += 1.0;
1396 let flipped_fp = NumericFeatureBuffer::from_f64_column_matrix(flipped)
1397 .unwrap()
1398 .fingerprint()
1399 .unwrap();
1400 assert_ne!(base_fp, flipped_fp);
1401 }
1402
1403 #[test]
1404 fn fingerprint_changes_when_a_feature_is_renamed() {
1405 let baseline = f64_column_matrix();
1406 let base_fp = NumericFeatureBuffer::from_f64_column_matrix(baseline.clone())
1407 .unwrap()
1408 .fingerprint()
1409 .unwrap();
1410 let mut renamed = baseline;
1411 renamed.feature_names[0] = "f0_renamed".to_string();
1412 let renamed_fp = NumericFeatureBuffer::from_f64_column_matrix(renamed)
1413 .unwrap()
1414 .fingerprint()
1415 .unwrap();
1416 assert_ne!(base_fp, renamed_fp);
1417 }
1418
1419 #[test]
1420 fn fingerprint_changes_when_observation_ids_are_reordered() {
1421 let baseline = f64_column_matrix();
1424 let base_fp = NumericFeatureBuffer::from_f64_column_matrix(baseline.clone())
1425 .unwrap()
1426 .fingerprint()
1427 .unwrap();
1428 let mut reordered = baseline.clone();
1429 reordered.observation_ids.swap(0, 2);
1430 for column in &mut reordered.columns {
1431 column.swap(0, 2);
1432 }
1433 if let Some(masks) = reordered.validity_masks.as_mut() {
1434 for mask in masks {
1435 mask.swap(0, 2);
1436 }
1437 }
1438 let reordered_fp = NumericFeatureBuffer::from_f64_column_matrix(reordered)
1439 .unwrap()
1440 .fingerprint()
1441 .unwrap();
1442 assert_ne!(base_fp, reordered_fp);
1443 }
1444
1445 #[test]
1446 fn fingerprint_distinguishes_transposed_shapes_with_identical_flat_values() {
1447 let two_by_three = dense_buffer(2, 3).fingerprint().unwrap();
1454 let three_by_two = dense_buffer(3, 2).fingerprint().unwrap();
1455 assert_ne!(two_by_three, three_by_two);
1456 }
1457
1458 #[test]
1459 fn shape_framing_alone_changes_digest() {
1460 let cells: [Option<f64>; 6] = [
1464 Some(0.0),
1465 Some(1.0),
1466 Some(2.0),
1467 Some(3.0),
1468 Some(4.0),
1469 Some(5.0),
1470 ];
1471 let digest = |rows: u64, cols: u64| {
1472 let mut hasher = StreamingHasher::new(b"shape-probe\0");
1473 hasher.absorb_u64(rows);
1474 hasher.absorb_u64(cols);
1475 for cell in cells {
1476 hasher.absorb_cell(cell);
1477 }
1478 hasher.finalize_hex()
1479 };
1480 assert_ne!(digest(2, 3), digest(3, 2));
1481 }
1482
1483 #[test]
1484 fn fingerprint_distinguishes_masked_cell_from_real_zero() {
1485 let masked = NumericFeatureMatrixF64Columnar {
1488 feature_set_id: "x".to_string(),
1489 representation_id: RepresentationId::new("tabular_numeric").unwrap(),
1490 feature_names: vec!["f0".to_string()],
1491 observation_ids: vec![oid("obs.0")],
1492 columns: vec![vec![0.0]],
1493 validity_masks: Some(vec![vec![false]]),
1494 };
1495 let real_zero = NumericFeatureMatrixF64Columnar {
1496 feature_set_id: "x".to_string(),
1497 representation_id: RepresentationId::new("tabular_numeric").unwrap(),
1498 feature_names: vec!["f0".to_string()],
1499 observation_ids: vec![oid("obs.0")],
1500 columns: vec![vec![0.0]],
1501 validity_masks: None,
1502 };
1503 let masked_fp = NumericFeatureBuffer::from_f64_column_matrix(masked)
1504 .unwrap()
1505 .fingerprint()
1506 .unwrap();
1507 let zero_fp = NumericFeatureBuffer::from_f64_column_matrix(real_zero)
1508 .unwrap()
1509 .fingerprint()
1510 .unwrap();
1511 assert_ne!(masked_fp, zero_fp);
1512 }
1513
1514 #[test]
1515 fn fingerprint_changes_when_representation_changes() {
1516 let baseline = f64_column_matrix();
1517 let base_fp = NumericFeatureBuffer::from_f64_column_matrix(baseline.clone())
1518 .unwrap()
1519 .fingerprint()
1520 .unwrap();
1521 let mut other = baseline;
1522 other.representation_id = RepresentationId::new("dense_signal").unwrap();
1523 let other_fp = NumericFeatureBuffer::from_f64_column_matrix(other)
1524 .unwrap()
1525 .fingerprint()
1526 .unwrap();
1527 assert_ne!(base_fp, other_fp);
1528 }
1529
1530 #[test]
1531 #[ignore = "perf sanity probe; run with --release --ignored --nocapture"]
1532 fn fingerprint_large_buffer_under_500ms() {
1533 let rows = 3021usize;
1541 let cols = 1050usize;
1542 let columns = (0..cols)
1543 .map(|c| {
1544 (0..rows)
1545 .map(|r| (r as f64) * 0.5 + (c as f64))
1546 .collect::<Vec<_>>()
1547 })
1548 .collect::<Vec<_>>();
1549 let matrix = NumericFeatureMatrixF64Columnar {
1550 feature_set_id: "big".to_string(),
1551 representation_id: RepresentationId::new("tabular_numeric").unwrap(),
1552 feature_names: (0..cols).map(|c| format!("f{c}")).collect(),
1553 observation_ids: (0..rows).map(|r| oid(&format!("obs.{r}"))).collect(),
1554 columns,
1555 validity_masks: None,
1556 };
1557 let buffer = NumericFeatureBuffer::from_f64_column_matrix(matrix).unwrap();
1558 let start = std::time::Instant::now();
1559 let fp = buffer.fingerprint().unwrap();
1560 let elapsed = start.elapsed();
1561 println!(
1562 "fingerprint({rows}x{cols}) = {:.3} ms (fp={fp})",
1563 elapsed.as_secs_f64() * 1e3
1564 );
1565 assert_eq!(fp.len(), 64);
1566 if !cfg!(debug_assertions) {
1567 assert!(
1568 elapsed.as_millis() < 500,
1569 "fingerprint took {} ms (>= 500 ms budget)",
1570 elapsed.as_millis()
1571 );
1572 }
1573 }
1574
1575 #[test]
1576 fn rejects_malformed_columnar_f64_matrix_shape() {
1577 let mut matrix = f64_column_matrix();
1578 matrix.columns[0].pop();
1579 let error = NumericFeatureBuffer::from_f64_column_matrix(matrix).unwrap_err();
1580 assert!(format!("{error}").contains("column 0 has 2 values"));
1581
1582 let mut matrix = f64_column_matrix();
1583 matrix.columns.pop();
1584 let error = NumericFeatureBuffer::from_f64_column_matrix(matrix).unwrap_err();
1585 assert!(format!("{error}").contains("has 1 columns for 2 features"));
1586
1587 let mut matrix = f64_column_matrix();
1588 matrix.validity_masks = Some(vec![vec![true, true, true]]);
1589 let error = NumericFeatureBuffer::from_f64_column_matrix(matrix).unwrap_err();
1590 assert!(format!("{error}").contains("has 1 validity_masks for 2 features"));
1591
1592 let mut matrix = f64_column_matrix();
1593 if let Some(masks) = matrix.validity_masks.as_mut() {
1594 masks[0].pop();
1595 }
1596 let error = NumericFeatureBuffer::from_f64_column_matrix(matrix).unwrap_err();
1597 assert!(format!("{error}").contains("column 0 validity_mask has 2 values"));
1598
1599 let mut matrix = f64_column_matrix();
1600 matrix.columns[0][0] = f64::NAN;
1601 let error = NumericFeatureBuffer::from_f64_column_matrix(matrix).unwrap_err();
1602 assert!(format!("{error}").contains("column 0 row 0 is not finite"));
1603
1604 let mut matrix = f64_column_matrix();
1605 matrix.columns[1][2] = f64::NAN;
1606 assert!(NumericFeatureBuffer::from_f64_column_matrix(matrix).is_ok());
1607 }
1608
1609 #[test]
1610 fn store_accepts_typed_f64_column_matrices() {
1611 let store =
1612 NumericFeatureBufferStore::from_f64_column_matrices(vec![f64_column_matrix()]).unwrap();
1613 let manifests = store.manifests().unwrap();
1614
1615 assert_eq!(manifests.len(), 1);
1616 assert_eq!(manifests[0].feature_set_id, "x");
1617 assert_eq!(manifests[0].value_count, 6);
1618
1619 let error = NumericFeatureBufferStore::from_f64_column_matrices(vec![
1620 f64_column_matrix(),
1621 f64_column_matrix(),
1622 ])
1623 .unwrap_err();
1624 assert!(format!("{error}").contains("duplicate f64 columnar feature matrix"));
1625 }
1626}