Transition to more specific stage 1 binning, compiles

disk/src/agg.rs

@@ -20,6 +20,7 @@ use std::time::{Duration, Instant};
 
 pub mod binnedt;
 pub mod binnedt2;
+pub mod binnedt3;
 pub mod binnedx;
 pub mod eventbatch;
 pub mod scalarbinbatch;

disk/src/agg/binnedt.rs

@@ -15,7 +15,7 @@ where
     SK: BinnedStreamKind,
 {
     type InputValue;
-    type OutputValue: AggregatableXdim1Bin<SK> + AggregatableTdim<SK> + Unpin;
+    type OutputValue;
     fn ends_before(&self, inp: &Self::InputValue) -> bool;
     fn ends_after(&self, inp: &Self::InputValue) -> bool;
     fn starts_after(&self, inp: &Self::InputValue) -> bool;
@@ -27,67 +27,63 @@ pub trait AggregatableTdim<SK>: Sized
 where
     SK: BinnedStreamKind,
 {
-    //type Output: AggregatableXdim1Bin + AggregatableTdim;
+    type Aggregator: AggregatorTdim<SK>;
-    type Aggregator: AggregatorTdim<SK, InputValue = Self, OutputValue = <SK as BinnedStreamKind>::TBinnedBins>;
     fn aggregator_new_static(ts1: u64, ts2: u64) -> Self::Aggregator;
 }
 
-pub trait IntoBinnedT<SK>
+pub trait IntoBinnedT<SK, S>
 where
     SK: BinnedStreamKind,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>> + Unpin,
 {
-    type StreamOut: Stream<
+    fn into_binned_t(self, spec: BinnedRange) -> IntoBinnedTDefaultStream<SK, S>;
-        Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>, Error>,
-    >;
-    fn into_binned_t(self, spec: BinnedRange) -> Self::StreamOut;
 }
 
-impl<S, I, SK> IntoBinnedT<SK> for S
+impl<SK, S> IntoBinnedT<SK, S> for S
 where
     SK: BinnedStreamKind,
-    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>> + Unpin,
-    I: AggregatableTdim<SK> + Unpin,
-    I::Aggregator: Unpin,
 {
-    type StreamOut = IntoBinnedTDefaultStream<S, I, SK>;
+    fn into_binned_t(self, spec: BinnedRange) -> IntoBinnedTDefaultStream<SK, S> {
-
-    fn into_binned_t(self, spec: BinnedRange) -> Self::StreamOut {
         IntoBinnedTDefaultStream::new(self, spec)
     }
 }
 
-pub struct IntoBinnedTDefaultStream<S, I, SK>
+pub struct IntoBinnedTDefaultStream<SK, S>
 where
-    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
-    I: AggregatableTdim<SK>,
     SK: BinnedStreamKind,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>> + Unpin,
 {
     inp: S,
-    aggtor: Option<I::Aggregator>,
+    aggtor: Option<<<SK as BinnedStreamKind>::XBinnedEvents as AggregatableTdim<SK>>::Aggregator>,
     spec: BinnedRange,
     curbin: u32,
     inp_completed: bool,
     all_bins_emitted: bool,
     range_complete_observed: bool,
     range_complete_emitted: bool,
-    left: Option<Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>>,
+    left:
+        Option<Poll<Option<Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>>>>,
     errored: bool,
     completed: bool,
-    tmp_agg_results: VecDeque<<I::Aggregator as AggregatorTdim<SK>>::OutputValue>,
+    tmp_agg_results: VecDeque<<SK as BinnedStreamKind>::TBinnedBins>,
     _marker: std::marker::PhantomData<SK>,
 }
 
-impl<S, I, SK> IntoBinnedTDefaultStream<S, I, SK>
+impl<SK, S> IntoBinnedTDefaultStream<SK, S>
 where
-    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
-    I: AggregatableTdim<SK>,
     SK: BinnedStreamKind,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>> + Unpin,
 {
     pub fn new(inp: S, spec: BinnedRange) -> Self {
         let range = spec.get_range(0);
         Self {
             inp,
-            aggtor: Some(I::aggregator_new_static(range.beg, range.end)),
+            aggtor: Some(
+                <<SK as BinnedStreamKind>::XBinnedEvents as AggregatableTdim<SK>>::aggregator_new_static(
+                    range.beg, range.end,
+                ),
+            ),
             spec,
             curbin: 0,
             inp_completed: false,
@@ -102,7 +98,10 @@ where
         }
     }
 
-    fn cur(&mut self, cx: &mut Context) -> Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>> {
+    fn cur(
+        &mut self,
+        cx: &mut Context,
+    ) -> Poll<Option<Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>>> {
         if let Some(cur) = self.left.take() {
             cur
         } else if self.inp_completed {
@@ -118,11 +117,16 @@ where
         let range = self.spec.get_range(self.curbin);
         let ret = self
             .aggtor
-            .replace(I::aggregator_new_static(range.beg, range.end))
+            .replace(
+                <<SK as BinnedStreamKind>::XBinnedEvents as AggregatableTdim<SK>>::aggregator_new_static(
+                    range.beg, range.end,
+                ),
+            )
             // TODO handle None case, or remove Option if Agg is always present
             .unwrap()
             .result();
-        self.tmp_agg_results = ret.into();
+        //self.tmp_agg_results = VecDeque::from(ret);
+        self.tmp_agg_results = VecDeque::new();
         if self.curbin >= self.spec.count as u32 {
             self.all_bins_emitted = true;
         }
@@ -130,12 +134,9 @@ where
 
     fn handle(
         &mut self,
-        cur: Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>,
+        cur: Poll<Option<Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>>>,
-    ) -> Option<
+    ) -> Option<Poll<Option<Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>, Error>>>>
-        Poll<
+    {
-            Option<Result<StreamItem<RangeCompletableItem<<I::Aggregator as AggregatorTdim<SK>>::OutputValue>>, Error>>,
-        >,
-    > {
         use Poll::*;
         match cur {
             Ready(Some(Ok(item))) => match item {
@@ -153,9 +154,11 @@ where
                             None
                         } else {
                             let ag = self.aggtor.as_mut().unwrap();
-                            if ag.ends_before(&item) {
+                            //if ag.ends_before(&item) {
+                            if ag.ends_before(err::todoval()) {
                                 None
-                            } else if ag.starts_after(&item) {
+                            //} else if ag.starts_after(&item) {
+                            } else if ag.starts_after(err::todoval()) {
                                 self.left =
                                     Some(Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))))));
                                 self.cycle_current_bin();
@@ -163,9 +166,11 @@ where
                                 None
                             } else {
                                 let mut item = item;
-                                ag.ingest(&mut item);
+                                //ag.ingest(&mut item);
+                                ag.ingest(err::todoval());
                                 let item = item;
-                                if ag.ends_after(&item) {
+                                //if ag.ends_after(&item) {
+                                if ag.ends_after(err::todoval()) {
                                     self.left =
                                         Some(Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))))));
                                     self.cycle_current_bin();
@@ -196,14 +201,11 @@ where
     }
 }
 
-impl<S, I, SK> Stream for IntoBinnedTDefaultStream<S, I, SK>
+impl<SK, S> Stream for IntoBinnedTDefaultStream<SK, S>
 where
-    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
-    I: AggregatableTdim<SK> + Unpin,
-    I::Aggregator: Unpin,
     SK: BinnedStreamKind,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>> + Unpin,
 {
-    //type Item = Result<StreamItem<RangeCompletableItem<<I::Aggregator as AggregatorTdim>::OutputValue>>, Error>;
     type Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {

disk/src/agg/binnedt3.rs

@@ -0,0 +1,101 @@
+use crate::agg::eventbatch::MinMaxAvgScalarEventBatch;
+use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
+use crate::agg::streams::StreamItem;
+use crate::binned::{BinnedStreamKind, RangeCompletableItem};
+use err::Error;
+use futures_core::Stream;
+use netpod::BinnedRange;
+use std::collections::VecDeque;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pub trait Aggregator3Tdim {
+    type InputValue;
+    type OutputValue;
+}
+
+pub struct Agg3 {}
+
+impl Aggregator3Tdim for Agg3 {
+    type InputValue = MinMaxAvgScalarEventBatch;
+    type OutputValue = MinMaxAvgScalarBinBatch;
+}
+
+pub struct BinnedT3Stream {
+    // TODO get rid of box:
+    inp: Pin<Box<dyn Stream<Item = MinMaxAvgScalarEventBatch> + Send>>,
+    //aggtor: Option<<<SK as BinnedStreamKind>::XBinnedEvents as AggregatableTdim<SK>>::Aggregator>,
+    aggtor: Option<()>,
+    spec: BinnedRange,
+    curbin: u32,
+    inp_completed: bool,
+    all_bins_emitted: bool,
+    range_complete_observed: bool,
+    range_complete_emitted: bool,
+    //left: Option<Poll<Option<Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>>>>,
+    left: Option<()>,
+    errored: bool,
+    completed: bool,
+    tmp_agg_results: VecDeque<MinMaxAvgScalarBinBatch>,
+}
+
+impl BinnedT3Stream {
+    pub fn new<S>(inp: S, spec: BinnedRange) -> Self
+    where
+        S: Stream<Item = MinMaxAvgScalarEventBatch> + Send + 'static,
+    {
+        // TODO simplify here, get rid of numeric parameter:
+        let range = spec.get_range(0);
+        Self {
+            inp: Box::pin(inp),
+            aggtor: None,
+            spec,
+            curbin: 0,
+            inp_completed: false,
+            all_bins_emitted: false,
+            range_complete_observed: false,
+            range_complete_emitted: false,
+            left: None,
+            errored: false,
+            completed: false,
+            tmp_agg_results: VecDeque::new(),
+        }
+    }
+}
+
+impl Stream for BinnedT3Stream {
+    type Item = Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarBinBatch>>, Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        use Poll::*;
+        'outer: loop {
+            break if self.completed {
+                panic!("IntoBinnedTDefaultStream  poll_next on completed");
+            } else if self.errored {
+                self.completed = true;
+                Ready(None)
+            } else if let Some(item) = self.tmp_agg_results.pop_front() {
+                Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))))
+            } else if self.range_complete_emitted {
+                self.completed = true;
+                Ready(None)
+            } else if self.inp_completed && self.all_bins_emitted {
+                self.range_complete_emitted = true;
+                if self.range_complete_observed {
+                    Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
+                } else {
+                    continue 'outer;
+                }
+            } else {
+                err::todo();
+                Pending
+                // TODO `cur` and `handle` are not yet taken over from binnedt.rs
+                /*let cur = self.cur(cx);
+                match self.handle(cur) {
+                    Some(item) => item,
+                    None => continue 'outer,
+                }*/
+            };
+        }
+    }
+}

disk/src/agg/eventbatch.rs

@@ -232,6 +232,12 @@ where
         } else {
             self.sum / self.sumc as f32
         };
+
+        // TODO impl problem:
+        // The return type of this function must be the concrete type that I implement for.
+        // Otherwise I have no chance building that values.
+        // I must somehow differently couple that to the SK.
+
         let v = MinMaxAvgScalarBinBatch {
             ts1s: vec![self.ts1],
             ts2s: vec![self.ts2],

disk/src/agg/streams.rs

@@ -36,103 +36,3 @@ pub trait ToJsonResult {
     type Output;
     fn to_json_result(&self) -> Result<Self::Output, Error>;
 }
-
-impl<T, SK> AggregatableXdim1Bin<SK> for StreamItem<T>
-where
-    SK: BinnedStreamKind,
-    T: AggregatableTdim<SK> + AggregatableXdim1Bin<SK>,
-{
-    type Output = StreamItem<<T as AggregatableXdim1Bin<SK>>::Output>;
-
-    fn into_agg(self) -> Self::Output {
-        match self {
-            Self::Log(item) => Self::Output::Log(item),
-            Self::Stats(item) => Self::Output::Stats(item),
-            Self::DataItem(item) => Self::Output::DataItem(item.into_agg()),
-        }
-    }
-}
-
-pub struct StreamItemAggregator<T, SK>
-where
-    T: AggregatableTdim<SK>,
-    SK: BinnedStreamKind,
-{
-    inner_agg: <T as AggregatableTdim<SK>>::Aggregator,
-}
-
-impl<T, SK> StreamItemAggregator<T, SK>
-where
-    T: AggregatableTdim<SK>,
-    SK: BinnedStreamKind,
-{
-    pub fn new(ts1: u64, ts2: u64) -> Self {
-        Self {
-            inner_agg: <T as AggregatableTdim<SK>>::aggregator_new_static(ts1, ts2),
-        }
-    }
-}
-
-impl<T, SK> AggregatorTdim<SK> for StreamItemAggregator<T, SK>
-where
-    T: AggregatableTdim<SK>,
-    SK: BinnedStreamKind,
-{
-    type InputValue = StreamItem<T>;
-    type OutputValue = StreamItem<<<T as AggregatableTdim<SK>>::Aggregator as AggregatorTdim<SK>>::OutputValue>;
-
-    fn ends_before(&self, inp: &Self::InputValue) -> bool {
-        match inp {
-            StreamItem::Log(_) => false,
-            StreamItem::Stats(_) => false,
-            StreamItem::DataItem(item) => self.inner_agg.ends_before(item),
-        }
-    }
-
-    fn ends_after(&self, inp: &Self::InputValue) -> bool {
-        match inp {
-            StreamItem::Log(_) => false,
-            StreamItem::Stats(_) => false,
-            StreamItem::DataItem(item) => self.inner_agg.ends_after(item),
-        }
-    }
-
-    fn starts_after(&self, inp: &Self::InputValue) -> bool {
-        match inp {
-            StreamItem::Log(_) => false,
-            StreamItem::Stats(_) => false,
-            StreamItem::DataItem(item) => self.inner_agg.starts_after(item),
-        }
-    }
-
-    fn ingest(&mut self, inp: &mut Self::InputValue) {
-        match inp {
-            StreamItem::Log(_) => {}
-            StreamItem::Stats(_) => {}
-            StreamItem::DataItem(item) => {
-                self.inner_agg.ingest(item);
-            }
-        }
-    }
-
-    fn result(self) -> Vec<Self::OutputValue> {
-        self.inner_agg
-            .result()
-            .into_iter()
-            .map(|k| StreamItem::DataItem(k))
-            .collect()
-    }
-}
-
-impl<T, SK> AggregatableTdim<SK> for StreamItem<T>
-where
-    T: AggregatableTdim<SK>,
-    SK: BinnedStreamKind,
-{
-    //type Output = StreamItem<<StreamItemAggregator<T> as AggregatorTdim>::OutputValue>;
-    type Aggregator = StreamItemAggregator<T, SK>;
-
-    fn aggregator_new_static(ts1: u64, ts2: u64) -> Self::Aggregator {
-        Self::Aggregator::new(ts1, ts2)
-    }
-}

disk/src/aggtest.rs

@@ -68,7 +68,7 @@ async fn agg_x_dim_0_inner() {
     );
     let fut1 = IntoDim1F32Stream::into_dim_1_f32_stream(fut1);
     let fut1 = IntoBinnedXBins1::<_, BinnedStreamKindScalar>::into_binned_x_bins_1(fut1);
-    let fut1 = IntoBinnedT::<BinnedStreamKindScalar>::into_binned_t(
+    let fut1 = IntoBinnedT::<BinnedStreamKindScalar, _>::into_binned_t(
         fut1,
         BinnedRange::covering_range(range, bin_count).unwrap().unwrap(),
     );
@@ -138,7 +138,7 @@ async fn agg_x_dim_1_inner() {
         //info!("after X binning  {:?}", k.as_ref().unwrap());
         k
     });
-    let fut1 = crate::agg::binnedt::IntoBinnedT::<BinnedStreamKindScalar>::into_binned_t(
+    let fut1 = crate::agg::binnedt::IntoBinnedT::<BinnedStreamKindScalar, _>::into_binned_t(
         fut1,
         BinnedRange::covering_range(range, bin_count).unwrap().unwrap(),
     );

disk/src/binned.rs

@@ -1,5 +1,6 @@
 use crate::agg::binnedt::{AggregatableTdim, AggregatorTdim, IntoBinnedT};
 use crate::agg::binnedt2::AggregatableTdim2;
+use crate::agg::binnedt3::{Agg3, BinnedT3Stream};
 use crate::agg::eventbatch::MinMaxAvgScalarEventBatch;
 use crate::agg::scalarbinbatch::{MinMaxAvgScalarBinBatch, MinMaxAvgScalarBinBatchAggregator};
 use crate::agg::streams::{Collectable, Collected, StreamItem, ToJsonResult};
@@ -77,6 +78,9 @@ impl Collected for MinMaxAvgScalarBinBatchCollected {
 pub struct MinMaxAvgScalarBinBatchCollectedJsonResult {
     ts_bin_edges: Vec<IsoDateTime>,
     counts: Vec<u64>,
+    mins: Vec<f32>,
+    maxs: Vec<f32>,
+    avgs: Vec<f32>,
     #[serde(skip_serializing_if = "Bool::is_false")]
     finalised_range: bool,
     #[serde(skip_serializing_if = "Zero::is_zero")]
@@ -108,6 +112,9 @@ impl ToJsonResult for MinMaxAvgScalarBinBatchCollected {
         };
         let ret = MinMaxAvgScalarBinBatchCollectedJsonResult {
             counts: self.batch.counts.clone(),
+            mins: self.batch.mins.clone(),
+            maxs: self.batch.maxs.clone(),
+            avgs: self.batch.avgs.clone(),
             missing_bins: self.bin_count_exp - self.batch.ts1s.len() as u32,
             finalised_range: self.finalised_range,
             ts_bin_edges: tsa,
@@ -531,6 +538,8 @@ pub trait BinnedStreamKind: Clone + Unpin + Send + Sync + 'static
         + 'static;
     type XBinnedEvents: XBinnedEvents<Self>;
     type TBinnedBins: TBinnedBins;
+    type XBinnedToTBinnedAggregator;
+    type XBinnedToTBinnedStream;
 
     fn new_binned_from_prebinned(
         &self,
@@ -547,6 +556,11 @@ pub trait BinnedStreamKind: Clone + Unpin + Send + Sync + 'static
         range: BinnedRange,
         node_config: &NodeConfigCached,
     ) -> Result<Self::TBinnedStreamType, Error>;
+
+    fn xbinned_to_tbinned<S>(inp: S, spec: BinnedRange) -> Self::XBinnedToTBinnedStream
+/*where
+        S: Stream<Item = Result<StreamItem<RangeCompletableItem<<Self as BinnedStreamKind>::XBinnedEvents>>, Error>>
+            + Unpin*/;
 }
 
 #[derive(Clone)]
@@ -578,6 +592,8 @@ impl BinnedStreamKind for BinnedStreamKindScalar {
     type TBinnedStreamType = BoxedStream<Result<StreamItem<RangeCompletableItem<Self::TBinnedBins>>, Error>>;
     type XBinnedEvents = MinMaxAvgScalarEventBatch;
     type TBinnedBins = MinMaxAvgScalarBinBatch;
+    type XBinnedToTBinnedAggregator = Agg3;
+    type XBinnedToTBinnedStream = BinnedT3Stream;
 
     fn new_binned_from_prebinned(
         &self,
@@ -607,10 +623,14 @@ impl BinnedStreamKind for BinnedStreamKindScalar {
         node_config: &NodeConfigCached,
     ) -> Result<Self::TBinnedStreamType, Error> {
         let s = MergedFromRemotes::new(evq, perf_opts, node_config.node_config.cluster.clone(), self.clone());
-        // TODO use the binned2 instead
+        let s = Self::xbinned_to_tbinned(s, range);
-        let s = crate::agg::binnedt::IntoBinnedT::<Self>::into_binned_t(s, range);
+        //let s = crate::agg::binnedt::IntoBinnedT::<Self>::into_binned_t(s, range);
         Ok(BoxedStream::new(Box::pin(s))?)
     }
+
+    fn xbinned_to_tbinned<S>(inp: S, spec: BinnedRange) -> Self::XBinnedToTBinnedStream {
+        err::todoval()
+    }
 }
 
 // TODO this code is needed somewhere:

disk/src/cache/pbv.rs

@@ -167,7 +167,7 @@ where
             self.node_config.node_config.cluster.clone(),
             self.stream_kind.clone(),
         );
-        let s1 = crate::agg::binnedt::IntoBinnedT::<SK>::into_binned_t(s1, range);
+        let s1 = crate::agg::binnedt::IntoBinnedT::<SK, _>::into_binned_t(s1, range);
         //self.fut2 = Some(Box::pin(s1));
         self.fut2 = err::todoval();
     }

disk/src/raw/bffr.rs

@@ -66,7 +66,7 @@ where
                                 >,
                             >(
                                 &frame,
-                                <<SK as BinnedStreamKind>::XBinnedEvents as XBinnedEvents>::frame_type(),
+                                <<SK as BinnedStreamKind>::XBinnedEvents as XBinnedEvents<SK>>::frame_type(),
                             ) {
                                 Ok(item) => match item {
                                     Ok(item) => Ready(Some(Ok(item))),