From a89e1cfc50f15533004dc4149b6f44f9d761dadf Mon Sep 17 00:00:00 2001
From: Dominik Werder <dominik.werder@gmail.com>
Date: Thu, 6 May 2021 18:25:45 +0200
Subject: [PATCH] When not going through pre-binned the read bytes are exact

---
 disk/src/binnedstream.rs | 1 +
 disk/src/cache/pbv.rs    | 5 ++++-
 disk/src/merge.rs        | 4 ----
 retrieval/src/test.rs    | 2 +-
 4 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/disk/src/binnedstream.rs b/disk/src/binnedstream.rs
index 57008a0..2b08c2b 100644
--- a/disk/src/binnedstream.rs
+++ b/disk/src/binnedstream.rs
@@ -65,6 +65,7 @@ impl BinnedStream {
                         }
                         Ok(PreBinnedItem::RangeComplete) => Some(Ok(MinMaxAvgScalarBinBatchStreamItem::RangeComplete)),
                         Ok(PreBinnedItem::EventDataReadStats(stats)) => {
+                            info!("BinnedStream  observes stats {:?}", stats);
                             Some(Ok(MinMaxAvgScalarBinBatchStreamItem::EventDataReadStats(stats)))
                         }
                         Ok(PreBinnedItem::Log(item)) => Some(Ok(MinMaxAvgScalarBinBatchStreamItem::Log(item))),
diff --git a/disk/src/cache/pbv.rs b/disk/src/cache/pbv.rs
index dd86f9b..56899fc 100644
--- a/disk/src/cache/pbv.rs
+++ b/disk/src/cache/pbv.rs
@@ -112,7 +112,10 @@ impl PreBinnedValueStream {
             match k {
                 Ok(Values(k)) => Ok(PreBinnedItem::Batch(k)),
                 Ok(RangeComplete) => Ok(PreBinnedItem::RangeComplete),
-                Ok(EventDataReadStats(stats)) => Ok(PreBinnedItem::EventDataReadStats(stats)),
+                Ok(EventDataReadStats(stats)) => {
+                    info!("PreBinnedValueStream  ✙ ✙ ✙ ✙ ✙ ✙ ✙ ✙ ✙ ✙ ✙ ✙ ✙    stats {:?}", stats);
+                    Ok(PreBinnedItem::EventDataReadStats(stats))
+                }
                 Ok(Log(item)) => Ok(PreBinnedItem::Log(item)),
                 Err(e) => Err(e),
             }
diff --git a/disk/src/merge.rs b/disk/src/merge.rs
index 6c028df..ee9003a 100644
--- a/disk/src/merge.rs
+++ b/disk/src/merge.rs
@@ -198,10 +198,6 @@ where
         }
     }
 
-    // This can:
-    // Do nothing if all have Val or Finished.
-    // But if some is None:
-    // We might get some Pending from upstream. In that case, caller also wants to abort here.
     fn replenish(self: &mut Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Error>> {
         use Poll::*;
         let mut pending = 0;
diff --git a/retrieval/src/test.rs b/retrieval/src/test.rs
index 224e6ba..af54009 100644
--- a/retrieval/src/test.rs
+++ b/retrieval/src/test.rs
@@ -50,7 +50,7 @@ async fn get_binned_0_inner() -> Result<(), Error> {
     get_binned_channel(
         "wave-f64-be-n21",
         "1970-01-01T00:20:10.000Z",
-        "1970-01-01T00:20:51.000Z",
+        "1970-01-01T00:20:15.000Z",
         4,
         &cluster,
     )