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