From f3d7bf9e094c9b2693eff84d9f26e9f350cc43d0 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Wed, 7 Jun 2023 16:15:28 +0200 Subject: [PATCH] rename the crate and fix some compile errors that I missed a while back --- Cargo.lock | 20 +-- .../Cargo.toml | 2 +- .../src/lib.rs | 19 +- libs/timeline_data_flow/src/tests.rs | 170 ++++++++++++++++++ 4 files changed, 194 insertions(+), 17 deletions(-) rename libs/{layer_map => timeline_data_flow}/Cargo.toml (91%) rename libs/{layer_map => timeline_data_flow}/src/lib.rs (96%) create mode 100644 libs/timeline_data_flow/src/tests.rs diff --git a/Cargo.lock b/Cargo.lock index bec1b9a9b6..6ec167c056 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1988,16 +1988,6 @@ dependencies = [ "libc", ] -[[package]] -name = "layer_map" -version = "0.1.0" -dependencies = [ - "thiserror", - "tokio", - "utils", - "workspace_hack", -] - [[package]] name = "lazy_static" version = "1.4.0" @@ -4001,6 +3991,16 @@ dependencies = [ "time-core", ] +[[package]] +name = "timeline_data_flow" +version = "0.1.0" +dependencies = [ + "thiserror", + "tokio", + "utils", + "workspace_hack", +] + [[package]] name = "tinytemplate" version = "1.2.1" diff --git a/libs/layer_map/Cargo.toml b/libs/timeline_data_flow/Cargo.toml similarity index 91% rename from libs/layer_map/Cargo.toml rename to libs/timeline_data_flow/Cargo.toml index 4e701ba871..fe50852764 100644 --- a/libs/layer_map/Cargo.toml +++ b/libs/timeline_data_flow/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "layer_map" +name = "timeline_data_flow" version = "0.1.0" edition.workspace = true license.workspace = true diff --git a/libs/layer_map/src/lib.rs b/libs/timeline_data_flow/src/lib.rs similarity index 96% rename from libs/layer_map/src/lib.rs rename to libs/timeline_data_flow/src/lib.rs index 04c0a5e185..e0c71bcaaa 100644 --- a/libs/layer_map/src/lib.rs +++ b/libs/timeline_data_flow/src/lib.rs @@ -108,7 +108,7 @@ pub trait Types { type HistoricLayer; type InMemoryLayer: InMemoryLayer + Clone; type HistoricStuff: HistoricStuff + Clone; - type GetReconstructPathError; + type GetReconstructPathError: std::error::Error; } /// Error returned by [`InMemoryLayer::put`]. @@ -238,13 +238,16 @@ impl Reader { /// See the [`crate`]-level docs for why we return [`ReconstructWork`] instead of a Page Image here. pub async fn get(&self, key: T::Key, lsn: T::Lsn) -> Result, GetError> { // XXX dedup with Writer::get_nowait - let state = self.wait.wait_for(lsn).await?; + let state = self.wait.wait_for(lsn).await.map_err(GetError::SeqWait)?; let inmem_records = state .inmem .as_ref() .map(|iml| iml.get(key, lsn)) .unwrap_or_default(); - let historic_path = state.historic.get_reconstruct_path(key, lsn)?; + let historic_path = state + .historic + .get_reconstruct_path(key, lsn) + .map_err(GetError::GetReconstructPath)?; Ok(ReconstructWork { key, lsn, @@ -287,7 +290,7 @@ impl Writer { lsn: T::Lsn, delta: T::DeltaRecord, ) -> Result<(), PutError> { - let (snapshot_lsn, snapshot) = self.advance.get_current_data(); + let (_snapshot_lsn, snapshot) = self.advance.get_current_data(); // TODO ensure snapshot_lsn <= lsn? let mut inmem = snapshot .inmem @@ -372,13 +375,17 @@ impl Writer { .advance .wait_for_timeout(lsn, Duration::from_secs(0)) // The await is never going to block because we pass from_secs(0). - .await?; + .await + .map_err(GetError::SeqWait)?; let inmem_records = state .inmem .as_ref() .map(|iml| iml.get(key, lsn)) .unwrap_or_default(); - let historic_path = state.historic.get_reconstruct_path(key, lsn)?; + let historic_path = state + .historic + .get_reconstruct_path(key, lsn) + .map_err(GetError::GetReconstructPath)?; Ok(ReconstructWork { key, lsn, diff --git a/libs/timeline_data_flow/src/tests.rs b/libs/timeline_data_flow/src/tests.rs new file mode 100644 index 0000000000..0ea69730d9 --- /dev/null +++ b/libs/timeline_data_flow/src/tests.rs @@ -0,0 +1,170 @@ +use std::collections::{btree_map::Entry, BTreeMap}; +use std::sync::Arc; +use utils::seqwait; + +/// The ZST for which we impl the `super::Types` type collection trait. +struct TestTypes; + +impl super::Types for TestTypes { + type Key = usize; + + type Lsn = usize; + + type LsnCounter = UsizeCounter; + + type DeltaRecord = &'static str; + + type HistoricLayer = Arc; + + type InMemoryLayer = TestInMemoryLayer; + + type HistoricStuff = TestHistoricStuff; +} + +/// For testing, our in-memory layer is a simple hashmap. +#[derive(Clone, Default, Debug)] +struct TestInMemoryLayer { + by_key: BTreeMap>, +} + +/// For testing, our historic layers are just in-memory layer objects with `frozen==true`. +struct TestHistoricLayer(TestInMemoryLayer); + +/// This is the data structure that impls the `HistoricStuff` trait. +#[derive(Default, Clone)] +struct TestHistoricStuff { + by_key: BTreeMap>>, +} + +/// `seqwait::MonotonicCounter` impl +#[derive(Copy, Clone)] +pub struct UsizeCounter(usize); + +// Our testing impl of HistoricStuff references the frozen InMemoryLayer objects +// from all the (key,lsn) entries that it covers. +// This mimics the (much more efficient) search tree in the real impl. +impl super::HistoricStuff for TestHistoricStuff { + type Types = TestTypes; + fn get_reconstruct_path( + &self, + key: usize, + lsn: usize, + ) -> Result>, super::GetReconstructPathError> { + let Some(bk) = self.by_key.get(&key) else { + return Ok(vec![]); + }; + Ok(bk.range(..=lsn).rev().map(|(_, l)| Arc::clone(l)).collect()) + } + + fn make_historic(&self, inmem: TestInMemoryLayer) -> Self { + // For the purposes of testing, just turn the inmemory layer historic through the type system + let historic = Arc::new(TestHistoricLayer(inmem)); + // Deep-copy + let mut copy = self.by_key.clone(); + // Add the references to `inmem` to the deep-copied struct + for (k, v) in historic.0.by_key.iter() { + for (lsn, _deltas) in v.into_iter() { + let by_key = copy.entry(*k).or_default(); + let overwritten = by_key.insert(*lsn, historic.clone()); + assert!(matches!(overwritten, None), "layers must not overlap"); + } + } + Self { by_key: copy } + } +} + +impl super::InMemoryLayer for TestInMemoryLayer { + type Types = TestTypes; + + fn put( + &mut self, + key: usize, + lsn: usize, + delta: &'static str, + ) -> Result> { + let mut clone = self.clone(); + drop(self); + let by_key = clone.by_key.entry(key).or_default(); + match by_key.entry(lsn) { + Entry::Occupied(_record) => { + return Err(super::InMemoryLayerPutError { + delta, + kind: super::InMemoryLayerPutErrorKind::AlreadyHaveRecordForKeyAndLsn, + }); + } + Entry::Vacant(vacant) => vacant.insert(delta), + }; + Ok(clone) + } + + fn get(&self, key: usize, lsn: usize) -> Vec<&'static str> { + let by_key = match self.by_key.get(&key) { + Some(by_key) => by_key, + None => return vec![], + }; + by_key + .range(..=lsn) + .map(|(_, v)| v) + .rev() + .cloned() + .collect() + } +} + +impl UsizeCounter { + pub fn new(inital: usize) -> Self { + UsizeCounter(inital) + } +} + +impl seqwait::MonotonicCounter for UsizeCounter { + fn cnt_advance(&mut self, new_val: usize) { + assert!(self.0 < new_val); + self.0 = new_val; + } + + fn cnt_value(&self) -> usize { + self.0 + } +} + +#[test] +fn basic() { + let lm = TestHistoricStuff::default(); + + let (r, mut rw) = super::new::(UsizeCounter::new(0), lm); + + let r = Arc::new(r); + let r2 = Arc::clone(&r); + + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + + let read_jh = rt.spawn(async move { r.get(0, 10).await }); + + let mut rw = rt.block_on(async move { + rw.put(0, 1, "foo").await.unwrap(); + rw.put(1, 1, "bar").await.unwrap(); + rw.put(0, 10, "baz").await.unwrap(); + rw + }); + + let read_res = rt.block_on(read_jh).unwrap().unwrap(); + assert!( + read_res.historic_path.is_empty(), + "we have pushed less than needed for flush" + ); + assert_eq!(read_res.inmem_records, vec!["baz", "foo"]); + + let rw = rt.block_on(async move { + rw.put(0, 11, "blup").await.unwrap(); + rw + }); + let read_res = rt.block_on(async move { r2.get(0, 11).await.unwrap() }); + assert_eq!(read_res.historic_path.len(), 0); + assert_eq!(read_res.inmem_records, vec!["blup", "baz", "foo"]); + + drop(rw); +}