4283961027

Build Type

push

github

Committed by GitHub

Commit Message

feat: Blue green cache (#1061)

Run Details

645 of 645 new or added lines in 45 files covered. (100.0%)

27779 of 39307 relevant lines covered (70.67%)

52489.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

66.67

/dozer-cache/src/cache/mod.rs

mod lmdb;
use std::fmt::Debug;

use self::expression::QueryExpression;
use crate::errors::CacheError;
use dozer_types::{
    node::SourceStates,
    serde::{Deserialize, Serialize},
    types::{IndexDefinition, Record, Schema, SchemaIdentifier},
};
pub use lmdb::cache_manager::{CacheManagerOptions, LmdbCacheManager};
pub mod expression;
pub mod index;
mod plan;
pub mod test_utils;

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(crate = "dozer_types::serde")]
pub struct RecordWithId {
    pub id: u64,
    pub record: Record,
}

impl RecordWithId {
    pub fn new(id: u64, record: Record) -> Self {
        Self { id, record }
    }
}

pub trait CacheManager: Send + Sync + Debug {
    /// Opens a cache in read-write mode with given name or an alias with that name.
    ///
    /// If the name is both an alias and a real name, it's treated as an alias.
    fn open_rw_cache(&self, name: &str) -> Result<Option<Box<dyn RwCache>>, CacheError>;

    /// Opens a cache in read-only mode with given name or an alias with that name.
    ///
    /// If the name is both an alias and a real name, it's treated as an alias.
    fn open_ro_cache(&self, name: &str) -> Result<Option<Box<dyn RoCache>>, CacheError>;

    /// Creates a new cache with given `schema`s, which can also be opened in read-only mode using `open_ro_cache`.
    ///
    /// Schemas cannot be changed after the cache is created.
    ///
    /// The cache's name is unique.
    fn create_cache(
        &self,
        schemas: Vec<(String, Schema, Vec<IndexDefinition>)>,
    ) -> Result<Box<dyn RwCache>, CacheError>;

    /// Creates an alias `alias` for a cache with name `name`.
    ///
    /// If `alias` already exists, it's overwritten. If cache with name `name` doesn't exist, the alias is still recorded.
    fn create_alias(&self, name: &str, alias: &str) -> Result<(), CacheError>;
}

pub trait RoCache: Send + Sync + Debug {
    /// Returns the name of the cache.
    fn name(&self) -> &str;

    // Schema Operations
    fn get_schema(&self, schema_identifier: SchemaIdentifier) -> Result<&Schema, CacheError>;
    fn get_schema_and_indexes_by_name(
        &self,
        name: &str,
    ) -> Result<&(Schema, Vec<IndexDefinition>), CacheError>;

    // Record Operations
    fn get(&self, key: &[u8]) -> Result<RecordWithId, CacheError>;
    fn count(&self, schema_name: &str, query: &QueryExpression) -> Result<usize, CacheError>;
    fn query(
        &self,
        schema_name: &str,
        query: &QueryExpression,
    ) -> Result<(&Schema, Vec<RecordWithId>), CacheError>;
}

pub trait RwCache: RoCache {
    // Record Operations
    /// Sets the version of the inserted record and inserts it into the cache. Returns the id of the newly inserted record.
    fn insert(&self, record: &mut Record) -> Result<u64, CacheError>;
    /// Returns version of the deleted record.
    fn delete(&self, key: &[u8]) -> Result<u32, CacheError>;
    /// Sets the version of the updated record and updates it in the cache. Returns the version of the record before the update.
    fn update(&self, key: &[u8], record: &mut Record) -> Result<u32, CacheError>;
    /// Commits the current transaction.
    fn commit(&self, checkpoint: &SourceStates) -> Result<(), CacheError>;
    /// Get the current checkpoint.
    fn get_checkpoint(&self) -> Result<SourceStates, CacheError>;
}

1	mod lmdb;
2	use std::fmt::Debug;
3
4	use self::expression::QueryExpression;
5	use crate::errors::CacheError;
6	use dozer_types::{
7	node::SourceStates,
8	serde::{Deserialize, Serialize},
9	types::{IndexDefinition, Record, Schema, SchemaIdentifier},
10	};
11	pub use lmdb::cache_manager::{CacheManagerOptions, LmdbCacheManager};
12	pub mod expression;
13	pub mod index;
14	mod plan;
15	pub mod test_utils;
16		×
17	#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]	63,862✔
18	#[serde(crate = "dozer_types::serde")]
19	pub struct RecordWithId {
20	pub id: u64,
21	pub record: Record,
22	}
23
24	impl RecordWithId {	×
25	pub fn new(id: u64, record: Record) -> Self {	1,133,906✔
26	Self { id, record }	1,133,906✔
27	}	1,133,906✔
28	}
29
30	pub trait CacheManager: Send + Sync + Debug {
31	/// Opens a cache in read-write mode with given name or an alias with that name.
32	///
33	/// If the name is both an alias and a real name, it's treated as an alias.
34	fn open_rw_cache(&self, name: &str) -> Result<Option<Box<dyn RwCache>>, CacheError>;
35
36	/// Opens a cache in read-only mode with given name or an alias with that name.
37	///
38	/// If the name is both an alias and a real name, it's treated as an alias.
39	fn open_ro_cache(&self, name: &str) -> Result<Option<Box<dyn RoCache>>, CacheError>;
40
41	/// Creates a new cache with given `schema`s, which can also be opened in read-only mode using `open_ro_cache`.
42	///
43	/// Schemas cannot be changed after the cache is created.
44	///
45	/// The cache's name is unique.
46	fn create_cache(
47	&self,
48	schemas: Vec<(String, Schema, Vec<IndexDefinition>)>,
49	) -> Result<Box<dyn RwCache>, CacheError>;
50
51	/// Creates an alias `alias` for a cache with name `name`.
52	///
53	/// If `alias` already exists, it's overwritten. If cache with name `name` doesn't exist, the alias is still recorded.
54	fn create_alias(&self, name: &str, alias: &str) -> Result<(), CacheError>;
55	}
56
57	pub trait RoCache: Send + Sync + Debug {
58	/// Returns the name of the cache.
59	fn name(&self) -> &str;
60
61	// Schema Operations
62	fn get_schema(&self, schema_identifier: SchemaIdentifier) -> Result<&Schema, CacheError>;
63	fn get_schema_and_indexes_by_name(
64	&self,
65	name: &str,
66	) -> Result<&(Schema, Vec<IndexDefinition>), CacheError>;
67
68	// Record Operations
69	fn get(&self, key: &[u8]) -> Result<RecordWithId, CacheError>;
70	fn count(&self, schema_name: &str, query: &QueryExpression) -> Result<usize, CacheError>;
71	fn query(
72	&self,
73	schema_name: &str,
74	query: &QueryExpression,
75	) -> Result<(&Schema, Vec<RecordWithId>), CacheError>;
76	}
77
78	pub trait RwCache: RoCache {
79	// Record Operations
80	/// Sets the version of the inserted record and inserts it into the cache. Returns the id of the newly inserted record.
81	fn insert(&self, record: &mut Record) -> Result<u64, CacheError>;
82	/// Returns version of the deleted record.
83	fn delete(&self, key: &[u8]) -> Result<u32, CacheError>;
84	/// Sets the version of the updated record and updates it in the cache. Returns the version of the record before the update.
85	fn update(&self, key: &[u8], record: &mut Record) -> Result<u32, CacheError>;
86	/// Commits the current transaction.
87	fn commit(&self, checkpoint: &SourceStates) -> Result<(), CacheError>;
88	/// Get the current checkpoint.
89	fn get_checkpoint(&self) -> Result<SourceStates, CacheError>;
90	}

getdozer / dozer / 4283961027

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous