Refactor (WIP) the event fetch pipeline

scyllaconn/src/bincache.rs

@@ -0,0 +1,468 @@
+use crate::errconv::ErrConv;
+use crate::events::EventsStreamScylla;
+use err::Error;
+use futures_util::{Future, Stream, StreamExt};
+use items_2::binsdim0::MinMaxAvgDim0Bins;
+use items_2::{empty_binned_dyn, empty_events_dyn, ChannelEvents, TimeBinned};
+use netpod::log::*;
+use netpod::query::{CacheUsage, PlainEventsQuery, RawEventsQuery};
+use netpod::timeunits::*;
+use netpod::{AggKind, ChannelTyped, ScalarType, Shape};
+use netpod::{PreBinnedPatchCoord, PreBinnedPatchIterator, PreBinnedPatchRange};
+use scylla::Session as ScySession;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use std::time::{Duration, Instant};
+
+pub async fn read_cached_scylla(
+    series: u64,
+    chn: &ChannelTyped,
+    coord: &PreBinnedPatchCoord,
+    _agg_kind: AggKind,
+    scy: &ScySession,
+) -> Result<Option<Box<dyn TimeBinned>>, Error> {
+    let vals = (
+        series as i64,
+        (coord.bin_t_len() / SEC) as i32,
+        (coord.patch_t_len() / SEC) as i32,
+        coord.ix() as i64,
+    );
+    let res = scy
+            .query_iter(
+                "select counts, avgs, mins, maxs from binned_scalar_f32 where series = ? and bin_len_sec = ? and patch_len_sec = ? and agg_kind = 'dummy-agg-kind' and offset = ?",
+                vals,
+            )
+            .await;
+    let mut res = res.err_conv().map_err(|e| {
+        error!("can not read from cache");
+        e
+    })?;
+    while let Some(item) = res.next().await {
+        let row = item.err_conv()?;
+        let edges = coord.edges();
+        let (counts, avgs, mins, maxs): (Vec<i64>, Vec<f32>, Vec<f32>, Vec<f32>) = row.into_typed().err_conv()?;
+        let mut counts_mismatch = false;
+        if edges.len() != counts.len() + 1 {
+            counts_mismatch = true;
+        }
+        if counts.len() != avgs.len() {
+            counts_mismatch = true;
+        }
+        let ts1s = edges[..(edges.len() - 1).min(edges.len())].to_vec();
+        let ts2s = edges[1.min(edges.len())..].to_vec();
+        if ts1s.len() != ts2s.len() {
+            error!("ts1s vs ts2s mismatch");
+            counts_mismatch = true;
+        }
+        if ts1s.len() != counts.len() {
+            counts_mismatch = true;
+        }
+        let avgs = avgs.into_iter().map(|x| x).collect::<Vec<_>>();
+        let mins = mins.into_iter().map(|x| x as _).collect::<Vec<_>>();
+        let maxs = maxs.into_iter().map(|x| x as _).collect::<Vec<_>>();
+        if counts_mismatch {
+            error!(
+                "mismatch:  edges {}  ts1s {}  ts2s {}  counts {}  avgs {}  mins {}  maxs {}",
+                edges.len(),
+                ts1s.len(),
+                ts2s.len(),
+                counts.len(),
+                avgs.len(),
+                mins.len(),
+                maxs.len(),
+            );
+        }
+        let counts: Vec<_> = counts.into_iter().map(|x| x as u64).collect();
+        // TODO construct a dyn TimeBinned using the scalar type and shape information.
+        // TODO place the values with little copying into the TimeBinned.
+        use ScalarType::*;
+        use Shape::*;
+        match &chn.shape {
+            Scalar => match &chn.scalar_type {
+                F64 => {
+                    let ret = MinMaxAvgDim0Bins::<f64> {
+                        ts1s,
+                        ts2s,
+                        counts,
+                        avgs,
+                        mins,
+                        maxs,
+                    };
+                    return Ok(Some(Box::new(ret)));
+                }
+                _ => {
+                    error!("TODO can not yet restore {:?} {:?}", chn.scalar_type, chn.shape);
+                    err::todoval()
+                }
+            },
+            _ => {
+                error!("TODO can not yet restore {:?} {:?}", chn.scalar_type, chn.shape);
+                err::todoval()
+            }
+        }
+    }
+    Ok(None)
+}
+
+#[allow(unused)]
+struct WriteFut<'a> {
+    chn: &'a ChannelTyped,
+    coord: &'a PreBinnedPatchCoord,
+    data: &'a dyn TimeBinned,
+    scy: &'a ScySession,
+}
+
+impl<'a> WriteFut<'a> {
+    #[allow(unused)]
+    fn new(
+        chn: &'a ChannelTyped,
+        coord: &'a PreBinnedPatchCoord,
+        data: &'a dyn TimeBinned,
+        scy: &'a ScySession,
+    ) -> Self {
+        Self { chn, coord, data, scy }
+    }
+}
+
+impl<'a> Future for WriteFut<'a> {
+    type Output = Result<(), Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        let _ = cx;
+        Poll::Ready(Ok(()))
+    }
+}
+
+pub fn write_cached_scylla<'a>(
+    series: u64,
+    _chn: &'a ChannelTyped,
+    coord: &'a PreBinnedPatchCoord,
+    //data: &'a dyn TimeBinned,
+    data: Box<dyn TimeBinned>,
+    //scy: &'a ScySession,
+    _scy: Arc<ScySession>,
+) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'a>> {
+    //let _chn = unsafe { &*(chn as *const ChannelTyped) };
+    //let data_ptr = data as *const dyn TimeBinned as usize;
+    //let scy_ptr = scy as *const ScySession as usize;
+    let fut = async move {
+        //let data = unsafe { &*(data_ptr as *const dyn TimeBinned) };
+        //let scy = unsafe { &*(scy_ptr as *const ScySession) };
+        let bin_len_sec = (coord.bin_t_len() / SEC) as i32;
+        let patch_len_sec = (coord.patch_t_len() / SEC) as i32;
+        let offset = coord.ix();
+        warn!(
+            "write_cached_scylla  len {}  where series = {} and bin_len_sec = {} and patch_len_sec = {} and agg_kind = 'dummy-agg-kind' and offset = {}",
+            data.counts().len(),
+            series,
+            bin_len_sec,
+            patch_len_sec,
+            offset,
+        );
+        let _data2 = data.counts().iter().map(|x| *x as i64).collect::<Vec<_>>();
+        /*
+        let stmt = scy.prepare("insert into binned_scalar_f32 (series, bin_len_sec, patch_len_sec, agg_kind, offset, counts, avgs, mins, maxs) values (?, ?, ?, 'dummy-agg-kind', ?, ?, ?, ?, ?)").await.err_conv()?;
+        scy.execute(
+            &stmt,
+            (
+                series as i64,
+                bin_len_sec,
+                patch_len_sec,
+                offset as i64,
+                data2,
+                data.avgs(),
+                data.mins(),
+                data.maxs(),
+            ),
+        )
+        .await
+        .err_conv()
+        .map_err(|e| {
+            error!("can not write to cache");
+            e
+        })?;
+        */
+        Ok(())
+    };
+    Box::pin(fut)
+}
+
+pub async fn fetch_uncached_data(
+    series: u64,
+    chn: ChannelTyped,
+    coord: PreBinnedPatchCoord,
+    agg_kind: AggKind,
+    cache_usage: CacheUsage,
+    scy: Arc<ScySession>,
+) -> Result<Option<(Box<dyn TimeBinned>, bool)>, Error> {
+    info!("fetch_uncached_data  {coord:?}");
+    // Try to find a higher resolution pre-binned grid which covers the requested patch.
+    let (bin, complete) = match PreBinnedPatchRange::covering_range(coord.patch_range(), coord.bin_count() + 1) {
+        Ok(Some(range)) => {
+            if coord.patch_range() != range.range() {
+                error!(
+                    "The chosen covering range does not exactly cover the requested patch  {:?}  vs  {:?}",
+                    coord.patch_range(),
+                    range.range()
+                );
+            }
+            fetch_uncached_higher_res_prebinned(
+                series,
+                &chn,
+                coord.clone(),
+                range,
+                agg_kind,
+                cache_usage.clone(),
+                scy.clone(),
+            )
+            .await
+        }
+        Ok(None) => fetch_uncached_binned_events(series, &chn, coord.clone(), agg_kind, scy.clone()).await,
+        Err(e) => Err(e),
+    }?;
+    if true || complete {
+        let edges = coord.edges();
+        if edges.len() < bin.len() + 1 {
+            error!(
+                "attempt to write overfull bin to cache  edges {}  bin {}",
+                edges.len(),
+                bin.len()
+            );
+            return Err(Error::with_msg_no_trace(format!(
+                "attempt to write overfull bin to cache"
+            )));
+        } else if edges.len() > bin.len() + 1 {
+            let missing = edges.len() - bin.len() - 1;
+            error!("attempt to write incomplete bin to cache missing {missing}");
+        }
+        if let CacheUsage::Use | CacheUsage::Recreate = &cache_usage {
+            // TODO pass data in safe way.
+            let _data = bin.as_ref();
+            //let fut = WriteFut::new(&chn, &coord, err::todoval(), &scy);
+            //fut.await?;
+            //write_cached_scylla(series, &chn, &coord, bin.as_ref(), &scy).await?;
+        }
+    }
+    Ok(Some((bin, complete)))
+}
+
+pub fn fetch_uncached_data_box(
+    series: u64,
+    chn: &ChannelTyped,
+    coord: &PreBinnedPatchCoord,
+    agg_kind: AggKind,
+    cache_usage: CacheUsage,
+    scy: Arc<ScySession>,
+) -> Pin<Box<dyn Future<Output = Result<Option<(Box<dyn TimeBinned>, bool)>, Error>> + Send>> {
+    Box::pin(fetch_uncached_data(
+        series,
+        chn.clone(),
+        coord.clone(),
+        agg_kind,
+        cache_usage,
+        scy,
+    ))
+}
+
+pub async fn fetch_uncached_higher_res_prebinned(
+    series: u64,
+    chn: &ChannelTyped,
+    coord: PreBinnedPatchCoord,
+    range: PreBinnedPatchRange,
+    agg_kind: AggKind,
+    cache_usage: CacheUsage,
+    scy: Arc<ScySession>,
+) -> Result<(Box<dyn TimeBinned>, bool), Error> {
+    let edges = coord.edges();
+    // TODO refine the AggKind scheme or introduce a new BinningOpts type and get time-weight from there.
+    let do_time_weight = true;
+    // We must produce some result with correct types even if upstream delivers nothing at all.
+    let bin0 = empty_binned_dyn(&chn.scalar_type, &chn.shape, &agg_kind);
+    let mut time_binner = bin0.time_binner_new(edges.clone(), do_time_weight);
+    let mut complete = true;
+    let patch_it = PreBinnedPatchIterator::from_range(range.clone());
+    for patch_coord in patch_it {
+        // We request data here for a Coord, meaning that we expect to receive multiple bins.
+        // The expectation is that we receive a single TimeBinned which contains all bins of that PatchCoord.
+        //let patch_coord = PreBinnedPatchCoord::new(patch.bin_t_len(), patch.patch_t_len(), patch.ix());
+        let (bin, comp) = pre_binned_value_stream_with_scy(
+            series,
+            chn,
+            &patch_coord,
+            agg_kind.clone(),
+            cache_usage.clone(),
+            scy.clone(),
+        )
+        .await?;
+        if let Err(msg) = bin.validate() {
+            error!(
+                "pre-binned intermediate issue {}  coord {:?}  patch_coord {:?}",
+                msg, coord, patch_coord
+            );
+        }
+        complete = complete && comp;
+        time_binner.ingest(bin.as_time_binnable_dyn());
+    }
+    // Fixed limit to defend against a malformed implementation:
+    let mut i = 0;
+    while i < 80000 && time_binner.bins_ready_count() < coord.bin_count() as usize {
+        let n1 = time_binner.bins_ready_count();
+        if false {
+            trace!(
+                "pre-binned extra cycle  {}  {}  {}",
+                i,
+                time_binner.bins_ready_count(),
+                coord.bin_count()
+            );
+        }
+        time_binner.cycle();
+        i += 1;
+        if time_binner.bins_ready_count() <= n1 {
+            warn!("pre-binned cycle did not add another bin, break");
+            break;
+        }
+    }
+    if time_binner.bins_ready_count() < coord.bin_count() as usize {
+        return Err(Error::with_msg_no_trace(format!(
+            "pre-binned unable to produce all bins for the patch  bins_ready {}  coord.bin_count {}  edges.len {}",
+            time_binner.bins_ready_count(),
+            coord.bin_count(),
+            edges.len(),
+        )));
+    }
+    let ready = time_binner
+        .bins_ready()
+        .ok_or_else(|| Error::with_msg_no_trace(format!("unable to produce any bins for the patch range")))?;
+    if let Err(msg) = ready.validate() {
+        error!("pre-binned final issue {}  coord {:?}", msg, coord);
+    }
+    Ok((ready, complete))
+}
+
+pub async fn fetch_uncached_binned_events(
+    series: u64,
+    chn: &ChannelTyped,
+    coord: PreBinnedPatchCoord,
+    agg_kind: AggKind,
+    scy: Arc<ScySession>,
+) -> Result<(Box<dyn TimeBinned>, bool), Error> {
+    let edges = coord.edges();
+    // TODO refine the AggKind scheme or introduce a new BinningOpts type and get time-weight from there.
+    let do_time_weight = true;
+    // We must produce some result with correct types even if upstream delivers nothing at all.
+    let bin0 = empty_events_dyn(&chn.scalar_type, &chn.shape, &agg_kind);
+    let mut time_binner = bin0.time_binner_new(edges.clone(), do_time_weight);
+    let deadline = Instant::now();
+    let deadline = deadline
+        .checked_add(Duration::from_millis(6000))
+        .ok_or_else(|| Error::with_msg_no_trace(format!("deadline overflow")))?;
+    let _evq = RawEventsQuery::new(chn.channel.clone(), coord.patch_range(), agg_kind);
+    let evq = PlainEventsQuery::new(chn.channel.clone(), coord.patch_range(), 4096, None, true);
+    let mut events_dyn = EventsStreamScylla::new(series, &evq, chn.scalar_type.clone(), chn.shape.clone(), scy, false);
+    let mut complete = false;
+    loop {
+        let item = tokio::time::timeout_at(deadline.into(), events_dyn.next()).await;
+        let item = match item {
+            Ok(Some(k)) => k,
+            Ok(None) => break,
+            Err(_) => {
+                error!("fetch_uncached_binned_events  timeout");
+                return Err(Error::with_msg_no_trace(format!(
+                    "TODO handle fetch_uncached_binned_events timeout"
+                )));
+            }
+        };
+        match item {
+            Ok(ChannelEvents::Events(item)) => {
+                time_binner.ingest(item.as_time_binnable_dyn());
+                // TODO could also ask the binner here whether we are "complete" to stop sending useless data.
+            }
+            Ok(ChannelEvents::Status(_)) => {
+                // TODO flag, should not happen.
+                return Err(Error::with_msg_no_trace(format!(
+                    "unexpected read of channel status events"
+                )));
+            }
+            Ok(ChannelEvents::RangeComplete) => {
+                complete = true;
+            }
+            Err(e) => return Err(e),
+        }
+    }
+    // Fixed limit to defend against a malformed implementation:
+    let mut i = 0;
+    while i < 80000 && time_binner.bins_ready_count() < coord.bin_count() as usize {
+        let n1 = time_binner.bins_ready_count();
+        if false {
+            trace!(
+                "events extra cycle {}  {}  {}",
+                i,
+                time_binner.bins_ready_count(),
+                coord.bin_count()
+            );
+        }
+        time_binner.cycle();
+        i += 1;
+        if time_binner.bins_ready_count() <= n1 {
+            warn!("events cycle did not add another bin, break");
+            break;
+        }
+    }
+    if time_binner.bins_ready_count() < coord.bin_count() as usize {
+        return Err(Error::with_msg_no_trace(format!(
+            "events unable to produce all bins for the patch  bins_ready {}  coord.bin_count {}  edges.len {}",
+            time_binner.bins_ready_count(),
+            coord.bin_count(),
+            edges.len(),
+        )));
+    }
+    let ready = time_binner
+        .bins_ready()
+        .ok_or_else(|| Error::with_msg_no_trace(format!("unable to produce any bins for the patch")))?;
+    if let Err(msg) = ready.validate() {
+        error!("time binned invalid {}  coord {:?}", msg, coord);
+    }
+    Ok((ready, complete))
+}
+
+pub async fn pre_binned_value_stream_with_scy(
+    series: u64,
+    chn: &ChannelTyped,
+    coord: &PreBinnedPatchCoord,
+    agg_kind: AggKind,
+    cache_usage: CacheUsage,
+    scy: Arc<ScySession>,
+) -> Result<(Box<dyn TimeBinned>, bool), Error> {
+    trace!("pre_binned_value_stream_with_scy  {chn:?}  {coord:?}");
+    if let (Some(item), CacheUsage::Use) = (
+        read_cached_scylla(series, chn, coord, agg_kind.clone(), &scy).await?,
+        &cache_usage,
+    ) {
+        info!("+++++++++++++    GOOD READ");
+        Ok((item, true))
+    } else {
+        if let CacheUsage::Use = &cache_usage {
+            warn!("--+--+--+--+--+--+   NOT YET CACHED");
+        }
+        let res = fetch_uncached_data_box(series, chn, coord, agg_kind, cache_usage, scy).await?;
+        let (bin, complete) =
+            res.ok_or_else(|| Error::with_msg_no_trace(format!("pre_binned_value_stream_with_scy got None bin")))?;
+        Ok((bin, complete))
+    }
+}
+
+pub async fn pre_binned_value_stream(
+    series: u64,
+    chn: &ChannelTyped,
+    coord: &PreBinnedPatchCoord,
+    agg_kind: AggKind,
+    cache_usage: CacheUsage,
+    scy: Arc<ScySession>,
+) -> Result<Pin<Box<dyn Stream<Item = Result<Box<dyn TimeBinned>, Error>> + Send>>, Error> {
+    trace!("pre_binned_value_stream  series {series}  {chn:?}  {coord:?}");
+    let res = pre_binned_value_stream_with_scy(series, chn, coord, agg_kind, cache_usage, scy).await?;
+    error!("TODO pre_binned_value_stream");
+    err::todo();
+    Ok(Box::pin(futures_util::stream::iter([Ok(res.0)])))
+}

scyllaconn/src/errconv.rs

@@ -0,0 +1,51 @@
+use err::Error;
+use scylla::frame::response::cql_to_rust::FromRowError as ScyFromRowError;
+use scylla::transport::errors::{NewSessionError as ScyNewSessionError, QueryError as ScyQueryError};
+
+pub trait ErrConv<T> {
+    fn err_conv(self) -> Result<T, Error>;
+}
+
+impl<T> ErrConv<T> for Result<T, tokio_postgres::Error> {
+    fn err_conv(self) -> Result<T, Error> {
+        match self {
+            Ok(k) => Ok(k),
+            Err(e) => Err(Error::with_msg(e.to_string())),
+        }
+    }
+}
+
+impl<T, A> ErrConv<T> for Result<T, async_channel::SendError<A>> {
+    fn err_conv(self) -> Result<T, Error> {
+        match self {
+            Ok(k) => Ok(k),
+            Err(e) => Err(Error::with_msg(e.to_string())),
+        }
+    }
+}
+impl<T> ErrConv<T> for Result<T, ScyQueryError> {
+    fn err_conv(self) -> Result<T, Error> {
+        match self {
+            Ok(k) => Ok(k),
+            Err(e) => Err(Error::with_msg_no_trace(format!("{e:?}"))),
+        }
+    }
+}
+
+impl<T> ErrConv<T> for Result<T, ScyNewSessionError> {
+    fn err_conv(self) -> Result<T, Error> {
+        match self {
+            Ok(k) => Ok(k),
+            Err(e) => Err(Error::with_msg_no_trace(format!("{e:?}"))),
+        }
+    }
+}
+
+impl<T> ErrConv<T> for Result<T, ScyFromRowError> {
+    fn err_conv(self) -> Result<T, Error> {
+        match self {
+            Ok(k) => Ok(k),
+            Err(e) => Err(Error::with_msg_no_trace(format!("{e:?}"))),
+        }
+    }
+}

scyllaconn/src/events.rs

@@ -0,0 +1,606 @@
+use crate::errconv::ErrConv;
+use err::Error;
+use futures_util::{Future, FutureExt, Stream, StreamExt};
+use items_2::eventsdim0::EventsDim0;
+use items_2::{ChannelEvents, Empty, Events};
+use netpod::log::*;
+use netpod::query::{ChannelStateEvents, PlainEventsQuery};
+use netpod::timeunits::*;
+use netpod::{Channel, Database, NanoRange, ScalarType, Shape};
+use scylla::Session as ScySession;
+use std::collections::VecDeque;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use tokio_postgres::Client as PgClient;
+
+macro_rules! read_values {
+    ($fname:ident, $self:expr, $ts_msp:expr) => {{
+        let fut = $fname($self.series, $ts_msp, $self.range.clone(), $self.fwd, $self.scy.clone());
+        let fut = fut.map(|x| {
+            match x {
+                Ok(k) => {
+                    // TODO why static needed?
+                    //let b = Box::new(k) as Box<dyn Events + 'static>;
+                    let b = Box::new(k) as Box<dyn Events>;
+                    Ok(b)
+                }
+                Err(e) => Err(e),
+            }
+        });
+        let fut = Box::pin(fut) as Pin<Box<dyn Future<Output = Result<Box<dyn Events>, Error>> + Send>>;
+        fut
+    }};
+}
+
+struct ReadValues {
+    series: i64,
+    scalar_type: ScalarType,
+    shape: Shape,
+    range: NanoRange,
+    ts_msps: VecDeque<u64>,
+    fwd: bool,
+    fut: Pin<Box<dyn Future<Output = Result<Box<dyn Events>, Error>> + Send>>,
+    scy: Arc<ScySession>,
+}
+
+impl ReadValues {
+    fn new(
+        series: i64,
+        scalar_type: ScalarType,
+        shape: Shape,
+        range: NanoRange,
+        ts_msps: VecDeque<u64>,
+        fwd: bool,
+        scy: Arc<ScySession>,
+    ) -> Self {
+        let mut ret = Self {
+            series,
+            scalar_type,
+            shape,
+            range,
+            ts_msps,
+            fwd,
+            fut: Box::pin(futures_util::future::ready(Err(Error::with_msg_no_trace(
+                "future not initialized",
+            )))),
+            scy,
+        };
+        ret.next();
+        ret
+    }
+
+    fn next(&mut self) -> bool {
+        if let Some(ts_msp) = self.ts_msps.pop_front() {
+            self.fut = self.make_fut(ts_msp, self.ts_msps.len() > 1);
+            true
+        } else {
+            false
+        }
+    }
+
+    fn make_fut(
+        &mut self,
+        ts_msp: u64,
+        _has_more_msp: bool,
+    ) -> Pin<Box<dyn Future<Output = Result<Box<dyn Events>, Error>> + Send>> {
+        let fut = match &self.shape {
+            Shape::Scalar => match &self.scalar_type {
+                ScalarType::I8 => {
+                    read_values!(read_next_values_scalar_i8, self, ts_msp)
+                }
+                ScalarType::I16 => {
+                    read_values!(read_next_values_scalar_i16, self, ts_msp)
+                }
+                ScalarType::I32 => {
+                    read_values!(read_next_values_scalar_i32, self, ts_msp)
+                }
+                ScalarType::F32 => {
+                    read_values!(read_next_values_scalar_f32, self, ts_msp)
+                }
+                ScalarType::F64 => {
+                    read_values!(read_next_values_scalar_f64, self, ts_msp)
+                }
+                _ => {
+                    error!("TODO ReadValues add more types");
+                    err::todoval()
+                }
+            },
+            Shape::Wave(_) => match &self.scalar_type {
+                ScalarType::U16 => {
+                    read_values!(read_next_values_array_u16, self, ts_msp)
+                }
+                _ => {
+                    error!("TODO ReadValues add more types");
+                    err::todoval()
+                }
+            },
+            _ => {
+                error!("TODO ReadValues add more types");
+                err::todoval()
+            }
+        };
+        fut
+    }
+}
+
+enum FrState {
+    New,
+    FindMsp(Pin<Box<dyn Future<Output = Result<VecDeque<u64>, Error>> + Send>>),
+    ReadBack1(ReadValues),
+    ReadBack2(ReadValues),
+    ReadValues(ReadValues),
+    Done,
+}
+
+pub struct EventsStreamScylla {
+    state: FrState,
+    series: u64,
+    scalar_type: ScalarType,
+    shape: Shape,
+    range: NanoRange,
+    ts_msps: VecDeque<u64>,
+    scy: Arc<ScySession>,
+    do_test_stream_error: bool,
+}
+
+impl EventsStreamScylla {
+    pub fn new(
+        series: u64,
+        evq: &PlainEventsQuery,
+        scalar_type: ScalarType,
+        shape: Shape,
+        scy: Arc<ScySession>,
+        do_test_stream_error: bool,
+    ) -> Self {
+        Self {
+            state: FrState::New,
+            series,
+            scalar_type,
+            shape,
+            range: evq.range().clone(),
+            ts_msps: VecDeque::new(),
+            scy,
+            do_test_stream_error,
+        }
+    }
+
+    fn ts_msps_found(&mut self, ts_msps: VecDeque<u64>) {
+        info!("found  ts_msps {ts_msps:?}");
+        self.ts_msps = ts_msps;
+        // Find the largest MSP which can potentially contain some event before the range.
+        let befores: Vec<_> = self
+            .ts_msps
+            .iter()
+            .map(|x| *x)
+            .filter(|x| *x < self.range.beg)
+            .collect();
+        if befores.len() >= 1 {
+            let st = ReadValues::new(
+                self.series as i64,
+                self.scalar_type.clone(),
+                self.shape.clone(),
+                self.range.clone(),
+                [befores[befores.len() - 1]].into(),
+                false,
+                self.scy.clone(),
+            );
+            self.state = FrState::ReadBack1(st);
+        } else if self.ts_msps.len() >= 1 {
+            let st = ReadValues::new(
+                self.series as i64,
+                self.scalar_type.clone(),
+                self.shape.clone(),
+                self.range.clone(),
+                self.ts_msps.clone(),
+                true,
+                self.scy.clone(),
+            );
+            self.state = FrState::ReadValues(st);
+        } else {
+            self.state = FrState::Done;
+        }
+    }
+
+    fn back_1_done(&mut self, item: Box<dyn Events>) -> Option<Box<dyn Events>> {
+        info!("back_1_done  len {}", item.len());
+        if item.len() == 0 {
+            // Find the 2nd largest MSP which can potentially contain some event before the range.
+            let befores: Vec<_> = self
+                .ts_msps
+                .iter()
+                .map(|x| *x)
+                .filter(|x| *x < self.range.beg)
+                .collect();
+            if befores.len() >= 2 {
+                let st = ReadValues::new(
+                    self.series as i64,
+                    self.scalar_type.clone(),
+                    self.shape.clone(),
+                    self.range.clone(),
+                    [befores[befores.len() - 2]].into(),
+                    false,
+                    self.scy.clone(),
+                );
+                self.state = FrState::ReadBack2(st);
+                None
+            } else if self.ts_msps.len() >= 1 {
+                let st = ReadValues::new(
+                    self.series as i64,
+                    self.scalar_type.clone(),
+                    self.shape.clone(),
+                    self.range.clone(),
+                    self.ts_msps.clone(),
+                    true,
+                    self.scy.clone(),
+                );
+                self.state = FrState::ReadValues(st);
+                None
+            } else {
+                self.state = FrState::Done;
+                None
+            }
+        } else {
+            if self.ts_msps.len() > 0 {
+                let st = ReadValues::new(
+                    self.series as i64,
+                    self.scalar_type.clone(),
+                    self.shape.clone(),
+                    self.range.clone(),
+                    self.ts_msps.clone(),
+                    true,
+                    self.scy.clone(),
+                );
+                self.state = FrState::ReadValues(st);
+                Some(item)
+            } else {
+                self.state = FrState::Done;
+                Some(item)
+            }
+        }
+    }
+
+    fn back_2_done(&mut self, item: Box<dyn Events>) -> Option<Box<dyn Events>> {
+        info!("back_2_done  len {}", item.len());
+        if self.ts_msps.len() >= 1 {
+            let st = ReadValues::new(
+                self.series as i64,
+                self.scalar_type.clone(),
+                self.shape.clone(),
+                self.range.clone(),
+                self.ts_msps.clone(),
+                true,
+                self.scy.clone(),
+            );
+            self.state = FrState::ReadValues(st);
+        } else {
+            self.state = FrState::Done;
+        }
+        if item.len() > 0 {
+            Some(item)
+        } else {
+            None
+        }
+    }
+}
+
+impl Stream for EventsStreamScylla {
+    type Item = Result<ChannelEvents, Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        use Poll::*;
+        if self.do_test_stream_error {
+            let e = Error::with_msg(format!("Test PRIVATE STREAM error."))
+                .add_public_msg(format!("Test PUBLIC STREAM error."));
+            return Ready(Some(Err(e)));
+        }
+        loop {
+            break match self.state {
+                FrState::New => {
+                    let fut = find_ts_msp(self.series as i64, self.range.clone(), self.scy.clone());
+                    let fut = Box::pin(fut);
+                    self.state = FrState::FindMsp(fut);
+                    continue;
+                }
+                FrState::FindMsp(ref mut fut) => match fut.poll_unpin(cx) {
+                    Ready(Ok(ts_msps)) => {
+                        self.ts_msps_found(ts_msps);
+                        continue;
+                    }
+                    Ready(Err(e)) => {
+                        self.state = FrState::Done;
+                        Ready(Some(Err(e)))
+                    }
+                    Pending => Pending,
+                },
+                FrState::ReadBack1(ref mut st) => match st.fut.poll_unpin(cx) {
+                    Ready(Ok(item)) => {
+                        if let Some(item) = self.back_1_done(item) {
+                            item.verify();
+                            item.output_info();
+                            Ready(Some(Ok(ChannelEvents::Events(item))))
+                        } else {
+                            continue;
+                        }
+                    }
+                    Ready(Err(e)) => {
+                        self.state = FrState::Done;
+                        Ready(Some(Err(e)))
+                    }
+                    Pending => Pending,
+                },
+                FrState::ReadBack2(ref mut st) => match st.fut.poll_unpin(cx) {
+                    Ready(Ok(item)) => {
+                        if let Some(item) = self.back_2_done(item) {
+                            item.verify();
+                            item.output_info();
+                            Ready(Some(Ok(ChannelEvents::Events(item))))
+                        } else {
+                            continue;
+                        }
+                    }
+                    Ready(Err(e)) => {
+                        self.state = FrState::Done;
+                        Ready(Some(Err(e)))
+                    }
+                    Pending => Pending,
+                },
+                FrState::ReadValues(ref mut st) => match st.fut.poll_unpin(cx) {
+                    Ready(Ok(item)) => {
+                        info!("read values");
+                        item.verify();
+                        item.output_info();
+                        if !st.next() {
+                            info!("ReadValues exhausted");
+                            self.state = FrState::Done;
+                        }
+                        Ready(Some(Ok(ChannelEvents::Events(item))))
+                    }
+                    Ready(Err(e)) => Ready(Some(Err(e))),
+                    Pending => Pending,
+                },
+                FrState::Done => Ready(None),
+            };
+        }
+    }
+}
+
+async fn find_series(channel: &Channel, pgclient: Arc<PgClient>) -> Result<(u64, ScalarType, Shape), Error> {
+    info!("find_series  channel {:?}", channel);
+    let rows = if let Some(series) = channel.series() {
+        let q = "select series, facility, channel, scalar_type, shape_dims from series_by_channel where series = $1";
+        pgclient.query(q, &[&(series as i64)]).await.err_conv()?
+    } else {
+        let q = "select series, facility, channel, scalar_type, shape_dims from series_by_channel where facility = $1 and channel = $2";
+        pgclient
+            .query(q, &[&channel.backend(), &channel.name()])
+            .await
+            .err_conv()?
+    };
+    if rows.len() < 1 {
+        return Err(Error::with_public_msg_no_trace(format!(
+            "No series found for {channel:?}"
+        )));
+    }
+    if rows.len() > 1 {
+        error!("Multiple series found for {channel:?}");
+        return Err(Error::with_public_msg_no_trace(
+            "Multiple series found for channel, can not return data for ambiguous series",
+        ));
+    }
+    let row = rows
+        .into_iter()
+        .next()
+        .ok_or_else(|| Error::with_public_msg_no_trace(format!("can not find series for channel")))?;
+    let series = row.get::<_, i64>(0) as u64;
+    let _facility: String = row.get(1);
+    let _channel: String = row.get(2);
+    let a: i32 = row.get(3);
+    let scalar_type = ScalarType::from_scylla_i32(a)?;
+    let a: Vec<i32> = row.get(4);
+    let shape = Shape::from_scylla_shape_dims(&a)?;
+    Ok((series, scalar_type, shape))
+}
+
+async fn find_ts_msp(series: i64, range: NanoRange, scy: Arc<ScySession>) -> Result<VecDeque<u64>, Error> {
+    trace!("find_ts_msp  series {}  {:?}", series, range);
+    // TODO use prepared statements
+    let cql = "select ts_msp from ts_msp where series = ? and ts_msp <= ? order by ts_msp desc limit 2";
+    let res = scy.query(cql, (series, range.beg as i64)).await.err_conv()?;
+    let mut before = VecDeque::new();
+    for row in res.rows_typed_or_empty::<(i64,)>() {
+        let row = row.err_conv()?;
+        before.push_front(row.0 as u64);
+    }
+    let cql = "select ts_msp from ts_msp where series = ? and ts_msp > ? and ts_msp < ?";
+    let res = scy
+        .query(cql, (series, range.beg as i64, range.end as i64))
+        .await
+        .err_conv()?;
+    let mut ret = VecDeque::new();
+    for h in before {
+        ret.push_back(h);
+    }
+    for row in res.rows_typed_or_empty::<(i64,)>() {
+        let row = row.err_conv()?;
+        ret.push_back(row.0 as u64);
+    }
+    trace!("found in total {} rows", ret.len());
+    Ok(ret)
+}
+
+macro_rules! read_next_scalar_values {
+    ($fname:ident, $st:ty, $scyty:ty, $table_name:expr) => {
+        async fn $fname(
+            series: i64,
+            ts_msp: u64,
+            range: NanoRange,
+            fwd: bool,
+            scy: Arc<ScySession>,
+        ) -> Result<EventsDim0<$st>, Error> {
+            type ST = $st;
+            type SCYTY = $scyty;
+            if ts_msp >= range.end {
+                warn!("given ts_msp {}  >=  range.end {}", ts_msp, range.end);
+            }
+            if range.end > i64::MAX as u64 {
+                return Err(Error::with_msg_no_trace(format!("range.end overflows i64")));
+            }
+            let res = if fwd {
+                let ts_lsp_min = if ts_msp < range.beg { range.beg - ts_msp } else { 0 };
+                let ts_lsp_max = if ts_msp < range.end { range.end - ts_msp } else { 0 };
+                trace!(
+                    "FWD  ts_msp {}  ts_lsp_min {}  ts_lsp_max {}  {}",
+                    ts_msp,
+                    ts_lsp_min,
+                    ts_lsp_max,
+                    stringify!($fname)
+                );
+                // TODO use prepared!
+                let cql = concat!(
+                    "select ts_lsp, pulse, value from ",
+                    $table_name,
+                    " where series = ? and ts_msp = ? and ts_lsp >= ? and ts_lsp < ?"
+                );
+                scy.query(cql, (series, ts_msp as i64, ts_lsp_min as i64, ts_lsp_max as i64))
+                    .await
+                    .err_conv()?
+            } else {
+                let ts_lsp_max = if ts_msp < range.beg { range.beg - ts_msp } else { 0 };
+                info!(
+                    "BCK  ts_msp {}  ts_lsp_max {}  range beg {}  end {}  {}",
+                    ts_msp,
+                    ts_lsp_max,
+                    range.beg,
+                    range.end,
+                    stringify!($fname)
+                );
+                // TODO use prepared!
+                let cql = concat!(
+                    "select ts_lsp, pulse, value from ",
+                    $table_name,
+                    " where series = ? and ts_msp = ? and ts_lsp < ? order by ts_lsp desc limit 1"
+                );
+                scy.query(cql, (series, ts_msp as i64, ts_lsp_max as i64))
+                    .await
+                    .err_conv()?
+            };
+            let mut ret = EventsDim0::<ST>::empty();
+            for row in res.rows_typed_or_empty::<(i64, i64, SCYTY)>() {
+                let row = row.err_conv()?;
+                let ts = ts_msp + row.0 as u64;
+                let pulse = row.1 as u64;
+                let value = row.2 as ST;
+                ret.push(ts, pulse, value);
+            }
+            trace!("found in total {} events  ts_msp {}", ret.tss.len(), ts_msp);
+            Ok(ret)
+        }
+    };
+}
+
+macro_rules! read_next_array_values {
+    ($fname:ident, $st:ty, $scyty:ty, $table_name:expr) => {
+        async fn $fname(
+            series: i64,
+            ts_msp: u64,
+            _range: NanoRange,
+            _fwd: bool,
+            scy: Arc<ScySession>,
+        ) -> Result<EventsDim0<$st>, Error> {
+            // TODO change return type: so far EventsDim1 does not exist.
+            error!("TODO read_next_array_values");
+            err::todo();
+            if true {
+                return Err(Error::with_msg_no_trace("redo based on scalar case"));
+            }
+            type ST = $st;
+            type _SCYTY = $scyty;
+            info!("{}  series {}  ts_msp {}", stringify!($fname), series, ts_msp);
+            let cql = concat!(
+                "select ts_lsp, pulse, value from ",
+                $table_name,
+                " where series = ? and ts_msp = ?"
+            );
+            let _res = scy.query(cql, (series, ts_msp as i64)).await.err_conv()?;
+            let ret = EventsDim0::<ST>::empty();
+            /*
+            for row in res.rows_typed_or_empty::<(i64, i64, Vec<SCYTY>)>() {
+                let row = row.err_conv()?;
+                let ts = ts_msp + row.0 as u64;
+                let pulse = row.1 as u64;
+                let value = row.2.into_iter().map(|x| x as ST).collect();
+                ret.push(ts, pulse, value);
+            }
+            */
+            info!("found in total {} events  ts_msp {}", ret.tss.len(), ts_msp);
+            Ok(ret)
+        }
+    };
+}
+
+read_next_scalar_values!(read_next_values_scalar_i8, i8, i8, "events_scalar_i8");
+read_next_scalar_values!(read_next_values_scalar_i16, i16, i16, "events_scalar_i16");
+read_next_scalar_values!(read_next_values_scalar_i32, i32, i32, "events_scalar_i32");
+read_next_scalar_values!(read_next_values_scalar_f32, f32, f32, "events_scalar_f32");
+read_next_scalar_values!(read_next_values_scalar_f64, f64, f64, "events_scalar_f64");
+
+read_next_array_values!(read_next_values_array_u16, u16, i16, "events_wave_u16");
+
+pub async fn make_scylla_stream(
+    evq: &PlainEventsQuery,
+    scy: Arc<ScySession>,
+    pgclient: &Arc<PgClient>,
+    do_test_stream_error: bool,
+) -> Result<Pin<Box<dyn Stream<Item = Result<ChannelEvents, Error>> + Send>>, Error> {
+    // Need to know details about the series in order to fetch data:
+    // TODO should query already contain ScalarType and Shape?
+    let (series, scalar_type, shape) = { find_series(evq.channel(), pgclient.clone()).await? };
+    let res = Box::pin(EventsStreamScylla::new(
+        series,
+        evq,
+        scalar_type,
+        shape,
+        scy,
+        do_test_stream_error,
+    )) as _;
+    Ok(res)
+}
+
+pub async fn channel_state_events(
+    evq: &ChannelStateEvents,
+    scy: Arc<ScySession>,
+    _dbconf: Database,
+) -> Result<Vec<(u64, u32)>, Error> {
+    let mut ret = Vec::new();
+    let div = DAY;
+    let mut ts_msp = evq.range().beg / div * div;
+    loop {
+        let series = (evq
+            .channel()
+            .series()
+            .ok_or(Error::with_msg_no_trace(format!("series id not given"))))?;
+        let params = (series as i64, ts_msp as i64);
+        let mut res = scy
+            .query_iter(
+                "select ts_lsp, kind from channel_status where series = ? and ts_msp = ?",
+                params,
+            )
+            .await
+            .err_conv()?;
+        while let Some(row) = res.next().await {
+            let row = row.err_conv()?;
+            let (ts_lsp, kind): (i64, i32) = row.into_typed().err_conv()?;
+            let ts = ts_msp + ts_lsp as u64;
+            let kind = kind as u32;
+            if ts >= evq.range().beg && ts < evq.range().end {
+                ret.push((ts, kind));
+            }
+        }
+        ts_msp += DAY;
+        if ts_msp >= evq.range().end {
+            break;
+        }
+    }
+    Ok(ret)
+}

scyllaconn/src/scyllaconn.rs

@@ -0,0 +1,20 @@
+pub mod bincache;
+pub mod errconv;
+pub mod events;
+
+use err::Error;
+use errconv::ErrConv;
+use netpod::ScyllaConfig;
+use scylla::Session as ScySession;
+use std::sync::Arc;
+
+pub async fn create_scy_session(scyconf: &ScyllaConfig) -> Result<Arc<ScySession>, Error> {
+    let scy = scylla::SessionBuilder::new()
+        .known_nodes(&scyconf.hosts)
+        .use_keyspace(&scyconf.keyspace, true)
+        .build()
+        .await
+        .err_conv()?;
+    let ret = Arc::new(scy);
+    Ok(ret)
+}