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.masks.iter().any(|mask| mask.present[idx]) {
546 return Err(DataError::Validation(format!(
547 "alignment sample `{sample_id}` is absent from every fused source"
548 )));
549 }
550 }
551 for block in blocks {
552 let mask = alignment
553 .masks
554 .iter()
555 .find(|mask| mask.source_id == block.source_id)
556 .expect("alignment sources were checked before presence validation");
557 let rows = row_maps
558 .get(&block.source_id)
559 .expect("source row map was created");
560 for (sample_id, present) in alignment.sample_ids.iter().zip(mask.present.iter()) {
561 let has_rows = rows.contains_key(sample_id);
562 if *present != has_rows {
563 return Err(DataError::Validation(format!(
564 "alignment presence for source `{}` sample `{sample_id}` is {present} but feature block rows are {}",
565 block.source_id,
566 if has_rows { "present" } else { "absent" }
567 )));
568 }
569 }
570 }
571 Ok(())
572}
573
574fn synthetic_observation_id(sample_id: &SampleId) -> Result<ObservationId> {
575 ObservationId::new(format!("fused.{}", sample_id.as_str()))
576}
577
578#[cfg(test)]
579mod tests {
580 use super::*;
581 use crate::alignment::{build_sample_alignment_plan, AlignmentMode, AlignmentPolicy};
582 use crate::ids::RepresentationId;
583 use serde_json::json;
584
585 fn block(
586 source_id: &str,
587 feature_names: &[&str],
588 rows: &[(&str, &str, Vec<serde_json::Value>)],
589 ) -> SourceFeatureBlock {
590 SourceFeatureBlock {
591 source_id: SourceId::new(source_id).unwrap(),
592 block: CoordinatorFeatureBlock {
593 feature_set_id: source_id.to_string(),
594 representation_id: RepresentationId::new("tabular_numeric").unwrap(),
595 feature_names: feature_names.iter().map(ToString::to_string).collect(),
596 observation_ids: rows
597 .iter()
598 .map(|(observation_id, _, _)| ObservationId::new(*observation_id).unwrap())
599 .collect(),
600 sample_ids: rows
601 .iter()
602 .map(|(_, sample_id, _)| SampleId::new(*sample_id).unwrap())
603 .collect(),
604 values: rows.iter().map(|(_, _, values)| values.clone()).collect(),
605 },
606 }
607 }
608
609 fn alignment(blocks: &[SourceFeatureBlock], mode: AlignmentMode) -> SampleAlignmentPlan {
610 let source_sets = blocks
611 .iter()
612 .map(source_sample_set_from_feature_block)
613 .collect::<Result<Vec<_>>>()
614 .unwrap();
615 build_sample_alignment_plan(&source_sets, &AlignmentPolicy { mode }).unwrap()
616 }
617
618 fn source_layout(
619 feature_set_id: &str,
620 blocks: &[SourceFeatureBlock],
621 ) -> FeatureFusionSourceLayout {
622 let mut column_start = 0;
623 let mut layout_blocks = Vec::new();
624 for block in blocks {
625 let column_count = block.block.feature_names.len();
626 layout_blocks.push(SourceFeatureLayoutBlock {
627 source_id: block.source_id.clone(),
628 preprocessing_output: SourcePreprocessingOutput {
629 feature_set_id: block.block.feature_set_id.clone(),
630 representation_id: block.block.representation_id.clone(),
631 adapter_id: Some(format!("preprocess_{}", block.source_id)),
632 fit_scope: Some(FitScope::FoldTrain),
633 },
634 column_start,
635 column_count,
636 feature_names: block.block.feature_names.clone(),
637 metadata: BTreeMap::new(),
638 });
639 column_start += column_count;
640 }
641
642 FeatureFusionSourceLayout {
643 kind: SourceFeatureLayoutKind::BySourceConcat,
644 source_order: blocks.iter().map(|block| block.source_id.clone()).collect(),
645 blocks: layout_blocks,
646 concat: SourceConcatLayout {
647 feature_set_id: feature_set_id.to_string(),
648 representation_id: blocks[0].block.representation_id.clone(),
649 axis: SourceConcatAxis::Feature,
650 total_column_count: column_start,
651 preserve_source_order: true,
652 namespace_columns: true,
653 },
654 metadata: BTreeMap::new(),
655 }
656 }
657
658 #[test]
659 fn source_layout_validates_source_order_and_concat_spans() {
660 let blocks = vec![
661 block(
662 "nir",
663 &["n0", "n1"],
664 &[("obs.S001.r1", "S001", vec![json!(1.0), json!(2.0)])],
665 ),
666 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
667 ];
668 let layout = source_layout("fused", &blocks);
669
670 layout.validate_for_source_blocks("fused", &blocks).unwrap();
671 assert_eq!(
672 layout.source_order,
673 vec![
674 SourceId::new("nir").unwrap(),
675 SourceId::new("chem").unwrap()
676 ]
677 );
678 assert_eq!(layout.blocks[0].column_start, 0);
679 assert_eq!(layout.blocks[1].column_start, 2);
680 assert_eq!(layout.concat.total_column_count, 3);
681 }
682
683 #[test]
684 fn source_layout_refuses_block_order_mismatch() {
685 let blocks = vec![
686 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
687 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
688 ];
689 let layout = source_layout("fused", &blocks);
690 let reversed = vec![blocks[1].clone(), blocks[0].clone()];
691
692 let error = layout
693 .validate_for_source_blocks("fused", &reversed)
694 .unwrap_err();
695
696 assert!(error.to_string().contains("feature fusion block"));
697 }
698
699 #[test]
700 fn source_layout_refuses_non_contiguous_concat_span() {
701 let blocks = vec![
702 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
703 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
704 ];
705 let mut layout = source_layout("fused", &blocks);
706 layout.blocks[1].column_start = 3;
707
708 let error = layout.validate().unwrap_err();
709
710 assert!(error.to_string().contains("expected contiguous start"));
711 }
712
713 #[test]
714 fn source_layout_refuses_preprocessing_output_mismatch() {
715 let blocks = vec![
716 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
717 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
718 ];
719 let mut layout = source_layout("fused", &blocks);
720 layout.blocks[0].preprocessing_output.feature_set_id = "other_nir_x".to_string();
721
722 let error = layout
723 .validate_for_source_blocks("fused", &blocks)
724 .unwrap_err();
725
726 assert!(error
727 .to_string()
728 .contains("preprocessing_output feature_set_id"));
729 }
730
731 #[test]
732 fn fusion_broadcasts_singleton_source_to_reference_repetitions() {
733 let blocks = vec![
734 block(
735 "nir",
736 &["n0"],
737 &[
738 ("obs.S001.r1", "S001", vec![json!(1.0)]),
739 ("obs.S001.r2", "S001", vec![json!(2.0)]),
740 ("obs.S002.r1", "S002", vec![json!(3.0)]),
741 ],
742 ),
743 block(
744 "chem",
745 &["c0"],
746 &[
747 ("chem.S001", "S001", vec![json!(10.0)]),
748 ("chem.S002", "S002", vec![json!(20.0)]),
749 ],
750 ),
751 ];
752 let fused = fuse_feature_blocks(
753 "fused",
754 &blocks,
755 &alignment(&blocks, AlignmentMode::Inner),
756 &FeatureFusionPolicy::default(),
757 )
758 .unwrap();
759
760 assert_eq!(fused.feature_names, vec!["nir.n0", "chem.c0"]);
761 assert_eq!(fused.sample_ids, blocks[0].block.sample_ids);
762 assert_eq!(
763 fused.values,
764 vec![
765 vec![json!(1.0), json!(10.0)],
766 vec![json!(2.0), json!(10.0)],
767 vec![json!(3.0), json!(20.0)]
768 ]
769 );
770 }
771
772 #[test]
773 fn outer_fusion_creates_synthetic_rows_for_samples_missing_in_reference() {
774 let blocks = vec![
775 block("nir", &["n0"], &[("obs.S001.r1", "S001", vec![json!(1.0)])]),
776 block(
777 "chem",
778 &["c0"],
779 &[
780 ("chem.S001", "S001", vec![json!(10.0)]),
781 ("chem.S002", "S002", vec![json!(20.0)]),
782 ],
783 ),
784 ];
785 let fused = fuse_feature_blocks(
786 "fused",
787 &blocks,
788 &alignment(&blocks, AlignmentMode::Outer),
789 &FeatureFusionPolicy::default(),
790 )
791 .unwrap();
792
793 assert_eq!(
794 fused
795 .observation_ids
796 .iter()
797 .map(ToString::to_string)
798 .collect::<Vec<_>>(),
799 vec!["obs.S001.r1", "fused.S002"]
800 );
801 assert_eq!(
802 fused.values,
803 vec![
804 vec![json!(1.0), json!(10.0)],
805 vec![serde_json::Value::Null, json!(20.0)]
806 ]
807 );
808 }
809
810 #[test]
811 fn fusion_refuses_ambiguous_non_reference_repetitions() {
812 let blocks = vec![
813 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
814 block(
815 "nir",
816 &["n0"],
817 &[
818 ("obs.S001.r1", "S001", vec![json!(1.0)]),
819 ("obs.S001.r2", "S001", vec![json!(2.0)]),
820 ],
821 ),
822 ];
823 let err = fuse_feature_blocks(
824 "fused",
825 &blocks,
826 &alignment(&blocks, AlignmentMode::Inner),
827 &FeatureFusionPolicy::default(),
828 )
829 .unwrap_err();
830
831 assert!(err.to_string().contains("cannot broadcast"));
832 }
833
834 #[test]
835 fn fusion_refuses_duplicate_unnamespaced_feature_names() {
836 let blocks = vec![
837 block("nir", &["x"], &[("nir.S001", "S001", vec![json!(1.0)])]),
838 block("chem", &["x"], &[("chem.S001", "S001", vec![json!(10.0)])]),
839 ];
840 let err = fuse_feature_blocks(
841 "fused",
842 &blocks,
843 &alignment(&blocks, AlignmentMode::Inner),
844 &FeatureFusionPolicy {
845 namespace_columns: false,
846 },
847 )
848 .unwrap_err();
849
850 assert!(err.to_string().contains("duplicate feature"));
851 }
852
853 #[test]
854 fn fusion_refuses_alignment_presence_that_does_not_match_rows() {
855 let blocks = vec![
856 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
857 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
858 ];
859 let mut bad_alignment = alignment(&blocks, AlignmentMode::Inner);
860 bad_alignment.masks[1].present[0] = false;
861 let err = fuse_feature_blocks(
862 "fused",
863 &blocks,
864 &bad_alignment,
865 &FeatureFusionPolicy::default(),
866 )
867 .unwrap_err();
868
869 assert!(err.to_string().contains("alignment presence"));
870 }
871
872 #[test]
873 fn fusion_refuses_alignment_sample_absent_from_all_sources() {
874 let blocks = vec![
875 block("nir", &["n0"], &[("nir.S001", "S001", vec![json!(1.0)])]),
876 block("chem", &["c0"], &[("chem.S001", "S001", vec![json!(10.0)])]),
877 ];
878 let mut bad_alignment = alignment(&blocks, AlignmentMode::Inner);
879 bad_alignment.masks[0].present[0] = false;
880 bad_alignment.masks[1].present[0] = false;
881 let err = fuse_feature_blocks(
882 "fused",
883 &blocks,
884 &bad_alignment,
885 &FeatureFusionPolicy::default(),
886 )
887 .unwrap_err();
888
889 assert!(err.to_string().contains("absent from every fused source"));
890 }
891}