1use std::collections::{BTreeMap, BTreeSet};
2
3use serde::{Deserialize, Serialize};
4
5use crate::alignment::{SampleAlignmentPlan, SourceSampleSet};
6use crate::error::{DataError, Result};
7use crate::handle::CoordinatorFeatureBlock;
8use crate::ids::{ObservationId, RepresentationId, SampleId, SourceId};
9use crate::plan::FitScope;
10
11#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
12pub struct FeatureFusionPolicy {
13 #[serde(default = "default_true")]
14 pub namespace_columns: bool,
15}
16
17impl Default for FeatureFusionPolicy {
18 fn default() -> Self {
19 Self {
20 namespace_columns: true,
21 }
22 }
23}
24
25fn default_true() -> bool {
26 true
27}
28
29#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
30pub struct SourceFeatureBlock {
31 pub source_id: SourceId,
32 pub block: CoordinatorFeatureBlock,
33}
34
35#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
36#[serde(rename_all = "snake_case")]
37pub enum SourceFeatureLayoutKind {
38 BySourceConcat,
39}
40
41#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
42#[serde(rename_all = "snake_case")]
43pub enum SourceConcatAxis {
44 Feature,
45}
46
47#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
48pub struct SourcePreprocessingOutput {
49 pub feature_set_id: String,
50 pub representation_id: RepresentationId,
51 #[serde(default, skip_serializing_if = "Option::is_none")]
52 pub adapter_id: Option<String>,
53 #[serde(default, skip_serializing_if = "Option::is_none")]
54 pub fit_scope: Option<FitScope>,
55}
56
57#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
58pub struct SourceFeatureLayoutBlock {
59 pub source_id: SourceId,
60 pub preprocessing_output: SourcePreprocessingOutput,
61 pub column_start: usize,
62 pub column_count: usize,
63 #[serde(default, skip_serializing_if = "Vec::is_empty")]
64 pub feature_names: Vec<String>,
65 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
66 pub metadata: BTreeMap<String, serde_json::Value>,
67}
68
69#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
70pub struct SourceConcatLayout {
71 pub feature_set_id: String,
72 pub representation_id: RepresentationId,
73 pub axis: SourceConcatAxis,
74 pub total_column_count: usize,
75 #[serde(default = "default_true")]
76 pub preserve_source_order: bool,
77 #[serde(default = "default_true")]
78 pub namespace_columns: bool,
79}
80
81#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
82pub struct FeatureFusionSourceLayout {
83 pub kind: SourceFeatureLayoutKind,
84 pub source_order: Vec<SourceId>,
85 pub blocks: Vec<SourceFeatureLayoutBlock>,
86 pub concat: SourceConcatLayout,
87 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
88 pub metadata: BTreeMap<String, serde_json::Value>,
89}
90
91impl SourcePreprocessingOutput {
92 pub fn validate(&self, label: &str) -> Result<()> {
93 if self.feature_set_id.trim().is_empty() {
94 return Err(DataError::Validation(format!(
95 "{label} preprocessing_output feature_set_id is empty"
96 )));
97 }
98 if self
99 .adapter_id
100 .as_ref()
101 .is_some_and(|adapter_id| adapter_id.trim().is_empty())
102 {
103 return Err(DataError::Validation(format!(
104 "{label} preprocessing_output adapter_id is empty"
105 )));
106 }
107 Ok(())
108 }
109}
110
111impl SourceFeatureLayoutBlock {
112 pub fn validate(&self, label: &str) -> Result<()> {
113 self.preprocessing_output.validate(label)?;
114 if self.column_count == 0 {
115 return Err(DataError::Validation(format!(
116 "{label} column_count must be greater than zero"
117 )));
118 }
119 if !self.feature_names.is_empty() && self.feature_names.len() != self.column_count {
120 return Err(DataError::Validation(format!(
121 "{label} feature_names length {} does not match column_count {}",
122 self.feature_names.len(),
123 self.column_count
124 )));
125 }
126 let mut seen = BTreeSet::new();
127 for feature_name in &self.feature_names {
128 if feature_name.trim().is_empty() {
129 return Err(DataError::Validation(format!(
130 "{label} contains an empty feature name"
131 )));
132 }
133 if !seen.insert(feature_name) {
134 return Err(DataError::Validation(format!(
135 "{label} contains duplicate feature `{feature_name}`"
136 )));
137 }
138 }
139 for key in self.metadata.keys() {
140 if key.trim().is_empty() {
141 return Err(DataError::Validation(format!(
142 "{label} metadata contains an empty key"
143 )));
144 }
145 }
146 Ok(())
147 }
148}
149
150impl SourceConcatLayout {
151 pub fn validate(&self) -> Result<()> {
152 if self.feature_set_id.trim().is_empty() {
153 return Err(DataError::Validation(
154 "source concat layout feature_set_id is empty".to_string(),
155 ));
156 }
157 if self.total_column_count == 0 {
158 return Err(DataError::Validation(
159 "source concat layout total_column_count must be greater than zero".to_string(),
160 ));
161 }
162 if !self.preserve_source_order {
163 return Err(DataError::Validation(
164 "source concat layout must preserve source order".to_string(),
165 ));
166 }
167 Ok(())
168 }
169}
170
171impl FeatureFusionSourceLayout {
172 pub fn validate(&self) -> Result<()> {
173 if self.source_order.is_empty() {
174 return Err(DataError::Validation(
175 "feature fusion source layout contains no source_order".to_string(),
176 ));
177 }
178 if self.blocks.len() != self.source_order.len() {
179 return Err(DataError::Validation(format!(
180 "feature fusion source layout has {} blocks for {} ordered sources",
181 self.blocks.len(),
182 self.source_order.len()
183 )));
184 }
185 self.concat.validate()?;
186
187 let mut seen_sources = BTreeSet::new();
188 let mut expected_column_start = 0usize;
189 for (idx, source_id) in self.source_order.iter().enumerate() {
190 if !seen_sources.insert(source_id) {
191 return Err(DataError::Validation(format!(
192 "feature fusion source layout contains duplicate source `{source_id}`"
193 )));
194 }
195 let block = &self.blocks[idx];
196 if &block.source_id != source_id {
197 return Err(DataError::Validation(format!(
198 "feature fusion source layout block {idx} is for `{}` but source_order has `{source_id}`",
199 block.source_id
200 )));
201 }
202 let label = format!("feature fusion source layout block `{}`", block.source_id);
203 block.validate(&label)?;
204 if block.column_start != expected_column_start {
205 return Err(DataError::Validation(format!(
206 "{label} starts at column {} but expected contiguous start {expected_column_start}",
207 block.column_start
208 )));
209 }
210 expected_column_start = expected_column_start
211 .checked_add(block.column_count)
212 .ok_or_else(|| {
213 DataError::Validation(
214 "feature fusion source layout column range overflows".to_string(),
215 )
216 })?;
217 }
218 if expected_column_start != self.concat.total_column_count {
219 return Err(DataError::Validation(format!(
220 "feature fusion source layout total_column_count {} does not match block span {}",
221 self.concat.total_column_count, expected_column_start
222 )));
223 }
224 for key in self.metadata.keys() {
225 if key.trim().is_empty() {
226 return Err(DataError::Validation(
227 "feature fusion source layout metadata contains an empty key".to_string(),
228 ));
229 }
230 }
231 Ok(())
232 }
233
234 pub fn validate_for_source_blocks(
235 &self,
236 feature_set_id: &str,
237 blocks: &[SourceFeatureBlock],
238 ) -> Result<()> {
239 self.validate()?;
240 if self.concat.feature_set_id != feature_set_id {
241 return Err(DataError::Validation(format!(
242 "source concat layout feature_set_id `{}` does not match requested fused feature_set_id `{feature_set_id}`",
243 self.concat.feature_set_id
244 )));
245 }
246 if blocks.len() != self.blocks.len() {
247 return Err(DataError::Validation(format!(
248 "source layout has {} blocks but feature fusion has {} source blocks",
249 self.blocks.len(),
250 blocks.len()
251 )));
252 }
253 for (idx, (layout_block, source_block)) in self.blocks.iter().zip(blocks.iter()).enumerate()
254 {
255 if layout_block.source_id != source_block.source_id {
256 return Err(DataError::Validation(format!(
257 "source layout block {idx} is for `{}` but feature fusion block is `{}`",
258 layout_block.source_id, source_block.source_id
259 )));
260 }
261 if layout_block.preprocessing_output.feature_set_id != source_block.block.feature_set_id
262 {
263 return Err(DataError::Validation(format!(
264 "source layout block `{}` preprocessing_output feature_set_id `{}` does not match feature block `{}`",
265 layout_block.source_id,
266 layout_block.preprocessing_output.feature_set_id,
267 source_block.block.feature_set_id
268 )));
269 }
270 if layout_block.preprocessing_output.representation_id
271 != source_block.block.representation_id
272 {
273 return Err(DataError::Validation(format!(
274 "source layout block `{}` preprocessing_output representation `{}` does not match feature block `{}`",
275 layout_block.source_id,
276 layout_block.preprocessing_output.representation_id,
277 source_block.block.representation_id
278 )));
279 }
280 if layout_block.column_count != source_block.block.feature_names.len() {
281 return Err(DataError::Validation(format!(
282 "source layout block `{}` column_count {} does not match feature block width {}",
283 layout_block.source_id,
284 layout_block.column_count,
285 source_block.block.feature_names.len()
286 )));
287 }
288 if !layout_block.feature_names.is_empty()
289 && layout_block.feature_names != source_block.block.feature_names
290 {
291 return Err(DataError::Validation(format!(
292 "source layout block `{}` feature_names do not match feature block output",
293 layout_block.source_id
294 )));
295 }
296 if self.concat.representation_id != source_block.block.representation_id {
297 return Err(DataError::Validation(format!(
298 "source concat layout representation `{}` does not match source `{}` representation `{}`",
299 self.concat.representation_id,
300 source_block.source_id,
301 source_block.block.representation_id
302 )));
303 }
304 }
305 Ok(())
306 }
307}
308
309pub fn source_sample_set_from_feature_block(block: &SourceFeatureBlock) -> Result<SourceSampleSet> {
310 validate_feature_block(&block.block)?;
311 let mut seen = BTreeSet::new();
312 let mut sample_ids = Vec::new();
313 for sample_id in &block.block.sample_ids {
314 if seen.insert(sample_id) {
315 sample_ids.push(sample_id.clone());
316 }
317 }
318 Ok(SourceSampleSet {
319 source_id: block.source_id.clone(),
320 sample_ids,
321 })
322}
323
324pub fn fuse_feature_blocks(
325 feature_set_id: impl Into<String>,
326 blocks: &[SourceFeatureBlock],
327 alignment: &SampleAlignmentPlan,
328 policy: &FeatureFusionPolicy,
329) -> Result<CoordinatorFeatureBlock> {
330 let feature_set_id = feature_set_id.into();
331 if feature_set_id.trim().is_empty() {
332 return Err(DataError::Validation(
333 "fused feature set id is empty".to_string(),
334 ));
335 }
336 if blocks.is_empty() {
337 return Err(DataError::Validation(
338 "feature fusion requires at least one source block".to_string(),
339 ));
340 }
341 alignment.validate()?;
342
343 let mut source_ids = BTreeSet::new();
344 for block in blocks {
345 if !source_ids.insert(&block.source_id) {
346 return Err(DataError::Validation(format!(
347 "feature fusion contains duplicate source `{}`",
348 block.source_id
349 )));
350 }
351 validate_feature_block(&block.block)?;
352 }
353 for mask in &alignment.masks {
354 if !source_ids.contains(&mask.source_id) {
355 return Err(DataError::Validation(format!(
356 "alignment mask references source `{}` absent from feature fusion",
357 mask.source_id
358 )));
359 }
360 }
361 if alignment.masks.len() != blocks.len() {
362 return Err(DataError::Validation(
363 "feature fusion sources and alignment masks differ".to_string(),
364 ));
365 }
366
367 let representation_id = blocks[0].block.representation_id.clone();
368 for block in blocks.iter().skip(1) {
369 if block.block.representation_id != representation_id {
370 return Err(DataError::Validation(format!(
371 "feature fusion source `{}` representation `{}` does not match `{}`",
372 block.source_id, block.block.representation_id, representation_id
373 )));
374 }
375 }
376
377 let mut feature_names = Vec::new();
378 for block in blocks {
379 for feature_name in &block.block.feature_names {
380 let output_name = if policy.namespace_columns {
381 format!("{}.{}", block.source_id, feature_name)
382 } else {
383 feature_name.clone()
384 };
385 feature_names.push(output_name);
386 }
387 }
388 let mut seen_features = BTreeSet::new();
389 for feature_name in &feature_names {
390 if !seen_features.insert(feature_name) {
391 return Err(DataError::Validation(format!(
392 "feature fusion produced duplicate feature `{feature_name}`"
393 )));
394 }
395 }
396
397 let row_maps = blocks
398 .iter()
399 .map(|block| (&block.source_id, rows_by_sample(&block.block)))
400 .collect::<BTreeMap<_, _>>();
401 validate_alignment_presence(blocks, alignment, &row_maps)?;
402 let reference = &blocks[0];
403 let reference_rows = row_maps
404 .get(&reference.source_id)
405 .expect("reference source map was created");
406
407 let mut observation_ids = Vec::new();
408 let mut sample_ids = Vec::new();
409 let mut values = Vec::new();
410
411 for sample_id in &alignment.sample_ids {
412 let output_rows = reference_rows
413 .get(sample_id)
414 .map(|indices| {
415 indices
416 .iter()
417 .map(|idx| OutputRow::Reference(*idx))
418 .collect::<Vec<_>>()
419 })
420 .unwrap_or_else(|| vec![OutputRow::Synthetic]);
421
422 for output_row in output_rows {
423 let mut row_values = Vec::new();
424 match output_row {
425 OutputRow::Reference(idx) => {
426 observation_ids.push(reference.block.observation_ids[idx].clone());
427 sample_ids.push(sample_id.clone());
428 }
429 OutputRow::Synthetic => {
430 observation_ids.push(synthetic_observation_id(sample_id)?);
431 sample_ids.push(sample_id.clone());
432 }
433 }
434 for block in blocks {
435 let source_rows = row_maps
436 .get(&block.source_id)
437 .expect("source row map was created");
438 if block.source_id == reference.source_id {
439 match output_row {
440 OutputRow::Reference(idx) => {
441 row_values.extend(block.block.values[idx].iter().cloned());
442 }
443 OutputRow::Synthetic => {
444 row_values.extend(std::iter::repeat_n(
445 serde_json::Value::Null,
446 block.block.feature_names.len(),
447 ));
448 }
449 }
450 continue;
451 }
452
453 match source_rows.get(sample_id).map(Vec::as_slice) {
454 Some([idx]) => row_values.extend(block.block.values[*idx].iter().cloned()),
455 Some(indices) => {
456 return Err(DataError::Validation(format!(
457 "feature fusion cannot broadcast {} repeated rows from non-reference source `{}` for sample `{sample_id}`",
458 indices.len(),
459 block.source_id
460 )));
461 }
462 None => row_values.extend(std::iter::repeat_n(
463 serde_json::Value::Null,
464 block.block.feature_names.len(),
465 )),
466 }
467 }
468 values.push(row_values);
469 }
470 }
471
472 let fused = CoordinatorFeatureBlock {
473 feature_set_id,
474 representation_id,
475 feature_names,
476 observation_ids,
477 sample_ids,
478 values,
479 };
480 validate_feature_block(&fused)?;
481 Ok(fused)
482}
483
484#[derive(Clone, Copy)]
485enum OutputRow {
486 Reference(usize),
487 Synthetic,
488}
489
490fn validate_feature_block(block: &CoordinatorFeatureBlock) -> Result<()> {
491 if block.feature_set_id.trim().is_empty() {
492 return Err(DataError::Validation(
493 "feature block feature_set_id is empty".to_string(),
494 ));
495 }
496 if block.feature_names.is_empty() {
497 return Err(DataError::Validation(format!(
498 "feature block `{}` contains no features",
499 block.feature_set_id
500 )));
501 }
502 if block.observation_ids.len() != block.sample_ids.len()
503 || block.sample_ids.len() != block.values.len()
504 {
505 return Err(DataError::Validation(format!(
506 "feature block `{}` row identity/value lengths differ",
507 block.feature_set_id
508 )));
509 }
510 let mut observations = BTreeSet::new();
511 for (idx, values) in block.values.iter().enumerate() {
512 if !observations.insert(&block.observation_ids[idx]) {
513 return Err(DataError::Validation(format!(
514 "feature block `{}` contains duplicate observation `{}`",
515 block.feature_set_id, block.observation_ids[idx]
516 )));
517 }
518 if values.len() != block.feature_names.len() {
519 return Err(DataError::Validation(format!(
520 "feature block `{}` row `{}` has {} values for {} features",
521 block.feature_set_id,
522 block.observation_ids[idx],
523 values.len(),
524 block.feature_names.len()
525 )));
526 }
527 }
528 Ok(())
529}
530
531fn rows_by_sample(block: &CoordinatorFeatureBlock) -> BTreeMap<&SampleId, Vec<usize>> {
532 let mut rows = BTreeMap::<&SampleId, Vec<usize>>::new();
533 for (idx, sample_id) in block.sample_ids.iter().enumerate() {
534 rows.entry(sample_id).or_default().push(idx);
535 }
536 rows
537}
538
539fn validate_alignment_presence<'a>(
540 blocks: &'a [SourceFeatureBlock],
541 alignment: &SampleAlignmentPlan,
542 row_maps: &BTreeMap<&'a SourceId, BTreeMap<&'a SampleId, Vec<usize>>>,
543) -> Result<()> {
544 for (idx, sample_id) in alignment.sample_ids.iter().enumerate() {
545 if alignment.mode == crate::alignment::AlignmentMode::Left
546 && !row_maps[&blocks[0].source_id].contains_key(sample_id)
547 {
548 return Err(DataError::Validation(format!(
549 "left alignment sample `{sample_id}` is absent from the reference source"
550 )));
551 }
552 if !alignment.masks.iter().any(|mask| mask.present[idx]) {
553 return Err(DataError::Validation(format!(
554 "alignment sample `{sample_id}` is absent from every fused source"
555 )));
556 }
557 }
558 for block in blocks {
559 let mask = alignment
560 .masks
561 .iter()
562 .find(|mask| mask.source_id == block.source_id)
563 .expect("alignment sources were checked before presence validation");
564 let rows = row_maps
565 .get(&block.source_id)
566 .expect("source row map was created");
567 for (sample_id, present) in alignment.sample_ids.iter().zip(mask.present.iter()) {
568 let has_rows = rows.contains_key(sample_id);
569 if *present != has_rows {
570 return Err(DataError::Validation(format!(
571 "alignment presence for source `{}` sample `{sample_id}` is {present} but feature block rows are {}",
572 block.source_id,
573 if has_rows { "present" } else { "absent" }
574 )));
575 }
576 }
577 }
578 Ok(())
579}
580
581fn synthetic_observation_id(sample_id: &SampleId) -> Result<ObservationId> {
582 ObservationId::new(format!("fused.{}", sample_id.as_str()))
583}
584
585#[cfg(test)]
586mod tests {
587 use super::*;
588 use crate::alignment::{build_sample_alignment_plan, AlignmentMode, AlignmentPolicy};
589 use crate::ids::RepresentationId;
590 use serde_json::json;
591
592 fn block(
593 source_id: &str,
594 feature_names: &[&str],
595 rows: &[(&str, &str, Vec<serde_json::Value>)],
596 ) -> SourceFeatureBlock {
597 SourceFeatureBlock {
598 source_id: SourceId::new(source_id).unwrap(),
599 block: CoordinatorFeatureBlock {
600 feature_set_id: source_id.to_string(),
601 representation_id: RepresentationId::new("tabular_numeric").unwrap(),
602 feature_names: feature_names.iter().map(ToString::to_string).collect(),
603 observation_ids: rows
604 .iter()
605 .map(|(observation_id, _, _)| ObservationId::new(*observation_id).unwrap())
606 .collect(),
607 sample_ids: rows
608 .iter()
609 .map(|(_, sample_id, _)| SampleId::new(*sample_id).unwrap())
610 .collect(),
611 values: rows.iter().map(|(_, _, values)| values.clone()).collect(),
612 },
613 }
614 }
615
616 fn alignment(blocks: &[SourceFeatureBlock], mode: AlignmentMode) -> SampleAlignmentPlan {
617 let source_sets = blocks
618 .iter()
619 .map(source_sample_set_from_feature_block)
620 .collect::<Result<Vec<_>>>()
621 .unwrap();
622 build_sample_alignment_plan(&source_sets, &AlignmentPolicy { mode }).unwrap()
623 }
624
625 fn source_layout(
626 feature_set_id: &str,
627 blocks: &[SourceFeatureBlock],
628 ) -> FeatureFusionSourceLayout {
629 let mut column_start = 0;
630 let mut layout_blocks = Vec::new();
631 for block in blocks {
632 let column_count = block.block.feature_names.len();
633 layout_blocks.push(SourceFeatureLayoutBlock {
634 source_id: block.source_id.clone(),
635 preprocessing_output: SourcePreprocessingOutput {
636 feature_set_id: block.block.feature_set_id.clone(),
637 representation_id: block.block.representation_id.clone(),
638 adapter_id: Some(format!("preprocess_{}", block.source_id)),
639 fit_scope: Some(FitScope::FoldTrain),
640 },
641 column_start,
642 column_count,
643 feature_names: block.block.feature_names.clone(),
644 metadata: BTreeMap::new(),
645 });
646 column_start += column_count;
647 }
648
649 FeatureFusionSourceLayout {
650 kind: SourceFeatureLayoutKind::BySourceConcat,
651 source_order: blocks.iter().map(|block| block.source_id.clone()).collect(),
652 blocks: layout_blocks,
653 concat: SourceConcatLayout {
654 feature_set_id: feature_set_id.to_string(),
655 representation_id: blocks[0].block.representation_id.clone(),
656 axis: SourceConcatAxis::Feature,
657 total_column_count: column_start,
658 preserve_source_order: true,
659 namespace_columns: true,
660 },
661 metadata: BTreeMap::new(),
662 }
663 }
664
665 #[test]
666 fn source_layout_validates_source_order_and_concat_spans() {
667 let blocks = vec![
668 block(
669 "nir",
670 &["n0", "n1"],
671 &[("obs.S001.r1", "S001", vec![json!(1.0), json!(2.0)])],
672 ),
673 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
674 ];
675 let layout = source_layout("fused", &blocks);
676
677 layout.validate_for_source_blocks("fused", &blocks).unwrap();
678 assert_eq!(
679 layout.source_order,
680 vec![
681 SourceId::new("nir").unwrap(),
682 SourceId::new("chem").unwrap()
683 ]
684 );
685 assert_eq!(layout.blocks[0].column_start, 0);
686 assert_eq!(layout.blocks[1].column_start, 2);
687 assert_eq!(layout.concat.total_column_count, 3);
688 }
689
690 #[test]
691 fn source_layout_refuses_block_order_mismatch() {
692 let blocks = vec![
693 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
694 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
695 ];
696 let layout = source_layout("fused", &blocks);
697 let reversed = vec![blocks[1].clone(), blocks[0].clone()];
698
699 let error = layout
700 .validate_for_source_blocks("fused", &reversed)
701 .unwrap_err();
702
703 assert!(error.to_string().contains("feature fusion block"));
704 }
705
706 #[test]
707 fn source_layout_refuses_non_contiguous_concat_span() {
708 let blocks = vec![
709 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
710 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
711 ];
712 let mut layout = source_layout("fused", &blocks);
713 layout.blocks[1].column_start = 3;
714
715 let error = layout.validate().unwrap_err();
716
717 assert!(error.to_string().contains("expected contiguous start"));
718 }
719
720 #[test]
721 fn source_layout_refuses_preprocessing_output_mismatch() {
722 let blocks = vec![
723 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
724 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
725 ];
726 let mut layout = source_layout("fused", &blocks);
727 layout.blocks[0].preprocessing_output.feature_set_id = "other_nir_x".to_string();
728
729 let error = layout
730 .validate_for_source_blocks("fused", &blocks)
731 .unwrap_err();
732
733 assert!(error
734 .to_string()
735 .contains("preprocessing_output feature_set_id"));
736 }
737
738 #[test]
739 fn fusion_broadcasts_singleton_source_to_reference_repetitions() {
740 let blocks = vec![
741 block(
742 "nir",
743 &["n0"],
744 &[
745 ("obs.S001.r1", "S001", vec![json!(1.0)]),
746 ("obs.S001.r2", "S001", vec![json!(2.0)]),
747 ("obs.S002.r1", "S002", vec![json!(3.0)]),
748 ],
749 ),
750 block(
751 "chem",
752 &["c0"],
753 &[
754 ("chem.S001", "S001", vec![json!(10.0)]),
755 ("chem.S002", "S002", vec![json!(20.0)]),
756 ],
757 ),
758 ];
759 let fused = fuse_feature_blocks(
760 "fused",
761 &blocks,
762 &alignment(&blocks, AlignmentMode::Inner),
763 &FeatureFusionPolicy::default(),
764 )
765 .unwrap();
766
767 assert_eq!(fused.feature_names, vec!["nir.n0", "chem.c0"]);
768 assert_eq!(fused.sample_ids, blocks[0].block.sample_ids);
769 assert_eq!(
770 fused.values,
771 vec![
772 vec![json!(1.0), json!(10.0)],
773 vec![json!(2.0), json!(10.0)],
774 vec![json!(3.0), json!(20.0)]
775 ]
776 );
777 }
778
779 #[test]
780 fn outer_fusion_creates_synthetic_rows_for_samples_missing_in_reference() {
781 let blocks = vec![
782 block("nir", &["n0"], &[("obs.S001.r1", "S001", vec![json!(1.0)])]),
783 block(
784 "chem",
785 &["c0"],
786 &[
787 ("chem.S001", "S001", vec![json!(10.0)]),
788 ("chem.S002", "S002", vec![json!(20.0)]),
789 ],
790 ),
791 ];
792 let fused = fuse_feature_blocks(
793 "fused",
794 &blocks,
795 &alignment(&blocks, AlignmentMode::Outer),
796 &FeatureFusionPolicy::default(),
797 )
798 .unwrap();
799
800 assert_eq!(
801 fused
802 .observation_ids
803 .iter()
804 .map(ToString::to_string)
805 .collect::<Vec<_>>(),
806 vec!["obs.S001.r1", "fused.S002"]
807 );
808 assert_eq!(
809 fused.values,
810 vec![
811 vec![json!(1.0), json!(10.0)],
812 vec![serde_json::Value::Null, json!(20.0)]
813 ]
814 );
815 }
816
817 #[test]
818 fn fusion_refuses_ambiguous_non_reference_repetitions() {
819 let blocks = vec![
820 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
821 block(
822 "nir",
823 &["n0"],
824 &[
825 ("obs.S001.r1", "S001", vec![json!(1.0)]),
826 ("obs.S001.r2", "S001", vec![json!(2.0)]),
827 ],
828 ),
829 ];
830 let err = fuse_feature_blocks(
831 "fused",
832 &blocks,
833 &alignment(&blocks, AlignmentMode::Inner),
834 &FeatureFusionPolicy::default(),
835 )
836 .unwrap_err();
837
838 assert!(err.to_string().contains("cannot broadcast"));
839 }
840
841 #[test]
842 fn fusion_refuses_duplicate_unnamespaced_feature_names() {
843 let blocks = vec![
844 block("nir", &["x"], &[("nir.S001", "S001", vec![json!(1.0)])]),
845 block("chem", &["x"], &[("chem.S001", "S001", vec![json!(10.0)])]),
846 ];
847 let err = fuse_feature_blocks(
848 "fused",
849 &blocks,
850 &alignment(&blocks, AlignmentMode::Inner),
851 &FeatureFusionPolicy {
852 namespace_columns: false,
853 },
854 )
855 .unwrap_err();
856
857 assert!(err.to_string().contains("duplicate feature"));
858 }
859
860 #[test]
861 fn fusion_refuses_alignment_presence_that_does_not_match_rows() {
862 let blocks = vec![
863 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
864 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
865 ];
866 let mut bad_alignment = alignment(&blocks, AlignmentMode::Inner);
867 bad_alignment.masks[1].present[0] = false;
868 let err = fuse_feature_blocks(
869 "fused",
870 &blocks,
871 &bad_alignment,
872 &FeatureFusionPolicy::default(),
873 )
874 .unwrap_err();
875
876 assert!(err.to_string().contains("alignment presence"));
877 }
878
879 #[test]
880 fn fusion_refuses_alignment_sample_absent_from_all_sources() {
881 let blocks = vec![
882 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
883 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
884 ];
885 let mut bad_alignment = alignment(&blocks, AlignmentMode::Inner);
886 bad_alignment.masks[0].present[0] = false;
887 bad_alignment.masks[1].present[0] = false;
888 let err = fuse_feature_blocks(
889 "fused",
890 &blocks,
891 &bad_alignment,
892 &FeatureFusionPolicy::default(),
893 )
894 .unwrap_err();
895
896 assert!(err.to_string().contains("absent from every fused source"));
897 }
898}