feat: Add par query and rm fold in agg trait

src/models/aggregators/raw_peak_agg/chromatogram_agg.rs

@@ -52,10 +52,6 @@ impl Aggregator<RawPeak> for ExtractedIonChromatomobilogram {
             .or_insert(u64_intensity);
     }
 
-    fn fold(&mut self, other: Self) {
-        panic!("Not Implemented;")
-    }
-
     fn finalize(self) -> ChromatomobilogramVectorArrayTuples {
         ChromatomobilogramVectorArrayTuples {
             scan_indices: self.scan_tree.into_iter().collect(),
@@ -72,6 +68,13 @@ impl Aggregator<RawPeak> for ExtractedIonChromatomobilogram {
 // type MappingCollection<T1, T2> = BTreeMap<T1, T2>;
 type MappingCollection<T1, T2> = HashMap<T1, T2>;
 
+#[derive(Debug, Clone)]
+pub struct MultiChromatomobilogramStats {
+    pub scan_tof_mapping:
+        MappingCollection<(usize, u32), (RunningStatsCalculator, RunningStatsCalculator)>,
+    pub id: u64,
+}
+
 #[derive(Debug, Clone)]
 pub struct ChromatomobilogramStats {
     // TODO OPTIMIZE THIS ... as needed.
@@ -107,7 +110,6 @@ impl Aggregator<RawPeak> for ChromatomobilogramStats {
         let u64_intensity = peak.intensity as u64;
         let rt_miliseconds = (peak.retention_time * 1000.0) as u32;
 
-        // TODO make this macro as well ...
         self.scan_tof_mapping
             .entry(rt_miliseconds)
             .and_modify(|curr| {
@@ -120,15 +122,9 @@ impl Aggregator<RawPeak> for ChromatomobilogramStats {
             ));
     }
 
-    fn fold(&mut self, _other: Self) {
-        panic!("Not Implemented;")
-    }
-
     fn finalize(self) -> ChromatomobilogramStatsArrays {
-        let ((scan_means, scan_sds), (tof_means, tof_sds)): (
-            (Vec<f64>, Vec<f64>),
-            (Vec<f64>, Vec<f64>),
-        ) = self
+        type VecTuple = (Vec<f64>, Vec<f64>);
+        let ((scan_means, scan_sds), (tof_means, tof_sds)): (VecTuple, VecTuple) = self
             .scan_tof_mapping
             .values()
             .map(|(scan, tof)| {

src/models/indices/raw_file_index.rs

@@ -167,10 +167,10 @@ impl RawFileIndex {
 }
 
 impl IndexedData<FragmentGroupIndexQuery, RawPeak> for RawFileIndex {
-    fn query(&self, fragment_query: &FragmentGroupIndexQuery) -> Option<Vec<RawPeak>> {
+    fn query(&self, fragment_query: &FragmentGroupIndexQuery) -> Vec<RawPeak> {
         let mut out = Vec::new();
         self.apply_on_query(fragment_query, &mut |peak, _| out.push(peak));
-        Some(out)
+        out
     }
 
     fn add_query<O, AG: crate::Aggregator<RawPeak, Output = O>>(

src/models/indices/transposed_quad_index/quad_splitted_transposed_index.rs

@@ -55,7 +55,6 @@ impl QuadSplittedTransposedIndex {
             rt_range,
             precursor_mz_range,
         );
-        // let matching_quads = self.get_matching_quad_settings(precursor_mz_range, scan_range);
         let matching_quads: Vec<SingleQuadrupoleSetting> = self
             .get_matching_quad_settings(precursor_mz_range, scan_range)
             .collect();
@@ -127,7 +126,8 @@ impl QuadSplittedTransposedIndex {
 
         assert!(frame_index_range.0 <= frame_index_range.1);
         assert!(mz_index_range.0 <= mz_index_range.1);
-        // assert!(mobility_index_range.0 <= mobility_index_range.1);
+        // Since mobilities get mixed up bc low scan ranges are high 1/k0, I
+        // Just make sure they are sorted here.
         let mobility_index_range = (
             mobility_index_range.0.min(mobility_index_range.1),
             mobility_index_range.1.max(mobility_index_range.0),
@@ -402,7 +402,7 @@ fn display_opt_peak_bucket_vec(opt_peak_buckets: &[Option<PeakBucket>]) -> Strin
 }
 
 impl IndexedData<FragmentGroupIndexQuery, RawPeak> for QuadSplittedTransposedIndex {
-    fn query(&self, fragment_query: &FragmentGroupIndexQuery) -> Option<Vec<RawPeak>> {
+    fn query(&self, fragment_query: &FragmentGroupIndexQuery) -> Vec<RawPeak> {
         let precursor_mz_range = (
             fragment_query.precursor_query.isolation_mz_range.0 as f64,
             fragment_query.precursor_query.isolation_mz_range.0 as f64,
@@ -412,7 +412,7 @@ impl IndexedData<FragmentGroupIndexQuery, RawPeak> for QuadSplittedTransposedInd
             fragment_query.precursor_query.frame_index_range,
         ));
 
-        let out = fragment_query
+        fragment_query
             .mz_index_ranges
             .iter()
             .flat_map(|tof_range| {
@@ -424,9 +424,7 @@ impl IndexedData<FragmentGroupIndexQuery, RawPeak> for QuadSplittedTransposedInd
                 )
                 .map(RawPeak::from)
             })
-            .collect();
-
-        Some(out)
+            .collect()
     }
 
     fn add_query<O, AG: crate::Aggregator<RawPeak, Output = O>>(
@@ -466,8 +464,8 @@ impl IndexedData<FragmentGroupIndexQuery, RawPeak> for QuadSplittedTransposedInd
         aggregator: &mut [AG],
     ) {
         fragment_queries
-            .iter()
-            .zip(aggregator.iter_mut())
+            .par_iter()
+            .zip(aggregator.par_iter_mut())
             .for_each(|(fragment_query, agg)| {
                 let precursor_mz_range = (
                     fragment_query.precursor_query.isolation_mz_range.0 as f64,

src/traits/aggregator.rs

@@ -2,6 +2,6 @@ pub trait Aggregator<I>: Send + Sync {
     type Output;
 
     fn add(&mut self, item: &I);
-    fn fold(&mut self, item: Self);
+    // fn fold(&mut self, item: Self);
     fn finalize(self) -> Self::Output;
 }

src/traits/indexed_data.rs

@@ -1,7 +1,7 @@
 use crate::Aggregator;
 
 pub trait IndexedData<QF, A> {
-    fn query(&self, fragment_query: &QF) -> Option<Vec<A>>;
+    fn query(&self, fragment_query: &QF) -> Vec<A>;
     fn add_query<O, AG: Aggregator<A, Output = O>>(&self, fragment_query: &QF, aggregator: &mut AG);
     fn add_query_multi_group<O, AG: Aggregator<A, Output = O>>(
         &self,