Skip to main content

dag_ml_data_core/
buffer.rs

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    /// Check the public descriptors against the immutable storage and index.
82    /// Callers may have edited a descriptor after construction.
83    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    /// Reproducibility fingerprint of the buffer's logical content as a 64-char
255    /// lowercase hex SHA-256.
256    ///
257    /// The preimage is fed to the hasher incrementally — there is NO
258    /// `O(n_cells)` intermediate byte vector or per-cell boxed value — so a
259    /// multi-million-cell matrix costs one fixed scratch buffer, not a
260    /// transient copy of the whole payload (the old `serde_json::to_vec` path).
261    ///
262    /// Preimage layout (every multi-byte field little-endian; every
263    /// variable-length field/collection length-prefixed so framing is
264    /// unambiguous):
265    ///
266    /// ```text
267    /// domain tag  b"dag-ml-data.numeric-feature-buffer.v2\0"   (fixed literal)
268    /// feature_set_id        : str   (u64 byte-len, then UTF-8)
269    /// representation_id     : str   (u64 byte-len, then UTF-8)
270    /// feature_names         : str[] (u64 count, then each str)
271    /// observation_ids       : str[] (u64 count, then each str)
272    /// n_rows                : u64   (fixed-width LE)
273    /// n_cols                : u64   (fixed-width LE)
274    /// cells row-major over (row, col):
275    ///   each cell           : 1 presence byte (1=Some, 0=None) + 8 LE f64 bytes
276    /// ```
277    ///
278    /// Rationale for the load-bearing choices:
279    /// * **Shape is hashed explicitly** as fixed-width `n_rows`/`n_cols`, and the
280    ///   cell stream is emitted in one fixed traversal, so a `2x3` and a `3x2`
281    ///   buffer with identical flat values produce different digests.
282    /// * **Storage is column-major** (`columns[feature][row]`); we deliberately
283    ///   traverse **row-major** over `(row, col)` and document that fixed order
284    ///   here. The constructors normalize both row-major and column-major typed
285    ///   inputs into the same `columns` storage, so the same logical matrix
286    ///   always streams in the same order regardless of input layout.
287    /// * **Masked cells** emit a `0` presence byte then 8 zero value bytes, a
288    ///   fixed 9-byte stride. The presence byte alone separates `None` from
289    ///   `Some(0.0)` (whose value bytes are also all zero), so a mask flip is
290    ///   never indistinguishable from a real value.
291    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    /// Returns the buffer's contents as a `NumericFeatureMatrixF64Columnar`,
326    /// suitable for binary persistence or pass-through to other typed-input
327    /// constructors. The output is the inverse of
328    /// `NumericFeatureBuffer::from_f64_column_matrix`: a round-trip through
329    /// this method and the constructor preserves the buffer fingerprint.
330    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    /// Like [`Self::project_relations`] but flattens the cells row-major
456    /// straight into one `Vec<f64>` — no `rows × cols` boxed
457    /// [`serde_json::Value`]s. This is the hot path for large numeric blocks
458    /// crossing a typed-array binding boundary. Masked (invalid) cells are an
459    /// error: the typed projection has no `Null` to project them into.
460    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    /// Deterministic iteration over `(feature_set_id, buffer)` pairs in
709    /// ascending `feature_set_id` order. Used by the file-store serializer
710    /// to produce byte-identical output for byte-identical stores.
711    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        // Unmasked buffer: project_relations_f64 must yield the boxed block's
1111        // rows concatenated row-major, with identical ids/names/ordering.
1112        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        // f64_matrix() masks obs.s2.nir/f1 — the boxed path projects Null, the
1138        // typed path must refuse (it has no Null to project into).
1139        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    /// Builds an `rows x cols` buffer whose flat row-major values are
1345    /// `0,1,2,...`, with generated feature / observation ids, no masking.
1346    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        // Reorder rows AND their values together so the buffer stays a valid
1422        // permutation of the same logical rows; only row order changes.
1423        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        // 2x3 and 3x2 share the flat row-major value sequence 0..6 but differ in
1448        // shape. At the buffer level this is over-determined (the feature-name
1449        // and observation-id collection counts also differ); the focused proof
1450        // that the explicit n_rows/n_cols framing alone is load-bearing is
1451        // `shape_framing_alone_changes_digest`, which holds the cell stream
1452        // constant and varies only the shape integers.
1453        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        // Reproduce just the shape + cell portion of the buffer preimage with
1461        // the SAME six cells but transposed (n_rows, n_cols), proving the shape
1462        // integers are what separate 2x3 from 3x2 — independent of names/ids.
1463        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        // None vs Some(0.0): the value bytes are identical (all zero); only the
1486        // presence byte separates them.
1487        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        // The historical serde_json path cost ~2.8 s on a 3021x1050 matrix in a
1534        // WASM context (a ~31 MB transient allocation). Optimized native must be
1535        // far under 500 ms with the streamed hash. The 500 ms budget is asserted
1536        // only in optimized builds: an unoptimized `cargo test` runs the sha2
1537        // compression with overflow checks and no SIMD, so its wall time is not
1538        // representative of "native" performance and would be a false failure.
1539        // The number is always printed.
1540        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}