4377467257

Build Type

push

github

Committed by GitHub

Commit Message

implement `HAVING` (#1198)

Run Details

395 of 395 new or added lines in 6 files covered. (100.0%)

27638 of 38584 relevant lines covered (71.63%)

27777.41 hits per line

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

98.04

/dozer-cache/src/cache/lmdb/utils.rs

use std::{fs, ops::Deref};

use crate::errors::CacheError;
use dozer_storage::{
    lmdb::EnvironmentFlags,
    lmdb_storage::{LmdbEnvironmentManager, LmdbEnvironmentOptions},
};
use tempdir::TempDir;

use super::cache::{CacheCommonOptions, CacheWriteOptions};

#[derive(Clone, Debug, Default)]
pub struct CacheOptions {
    pub common: CacheCommonOptions,
    pub kind: CacheOptionsKind,
}

#[derive(Clone, Debug, Default)]
pub struct CacheReadOptions {}

#[derive(Clone, Debug)]
pub enum CacheOptionsKind {
    // Write Options
    Write(CacheWriteOptions),

    // Read Options
    ReadOnly(CacheReadOptions),
}

impl Default for CacheOptionsKind {
    fn default() -> Self {
        Self::Write(CacheWriteOptions::default())
    }
}

pub fn init_env(options: &CacheOptions) -> Result<(LmdbEnvironmentManager, String), CacheError> {
    match &options.kind {
        CacheOptionsKind::Write(write_options) => {
            let (base_path, name, _temp_dir) = match &options.common.path {
                None => {
                    let base_path = TempDir::new("dozer")?;
                    (
                        base_path.path().to_path_buf(),
                        "dozer-cache",
                        Some(base_path),
                    )
                }
                Some((base_path, name)) => {
                    fs::create_dir_all(base_path)?;
                    (base_path.clone(), name.deref(), None)
                }
            };

            let options = LmdbEnvironmentOptions::new(
                options.common.max_db_size,
                options.common.max_readers,
                write_options.max_size,
                EnvironmentFlags::empty(),
            );

            Ok((
                LmdbEnvironmentManager::create(&base_path, name, options)?,
                name.to_string(),
            ))
        }
        CacheOptionsKind::ReadOnly(_) => {
            let (base_path, name) = options
                .common
                .path
                .as_ref()
                .ok_or(CacheError::PathNotInitialized)?;

            let env_options = LmdbEnvironmentOptions {
                max_dbs: options.common.max_db_size,
                max_readers: options.common.max_readers,
                flags: EnvironmentFlags::READ_ONLY,
                ..Default::default()
            };

            Ok((
                LmdbEnvironmentManager::create(base_path, name, env_options)?,
                name.to_string(),
            ))
        }
    }
}

#[cfg(test)]
mod tests {
    use dozer_storage::lmdb::{Cursor, DatabaseFlags, RoCursor, Transaction, WriteFlags};
    use dozer_types::types::Field;

    use crate::cache::lmdb::utils::{init_env, CacheOptions};

    fn cursor_dump(mut cursor: RoCursor) -> Vec<(&[u8], &[u8])> {
        cursor
            .iter_dup()
            .flatten()
            .collect::<dozer_storage::lmdb::Result<Vec<_>>>()
            .unwrap()
    }

    #[test]
    fn duplicate_test_nested() {
        let options = CacheOptions::default();
        let mut env = init_env(&options).unwrap().0;

        let db = env
            .create_database(
                Some("test"),
                Some(DatabaseFlags::DUP_SORT | DatabaseFlags::INTEGER_KEY),
            )
            .unwrap();

        let txn = env.create_txn().unwrap();
        let mut master_txn = txn.write();
        let txn = master_txn.txn_mut();

        let mut c_txn = txn.begin_nested_txn().unwrap();

        let items: Vec<(i64, &[u8])> = vec![
            (1, b"a"),
            (2, b"a"),
            (3, b"a"),
            (1, b"b"),
            (2, b"b"),
            (3, b"b"),
            (1, b"c"),
            (2, b"c"),
            (3, b"c"),
            (1, b"e"),
            (2, b"e"),
            (3, b"e"),
        ];
        for (key, data) in &items {
            let key = [
                "idx".as_bytes().to_vec(),
                Field::Int(*key).encode(),
                key.to_be_bytes().to_vec(),
            ]
            .join("#".as_bytes());
            c_txn.put(db, &key, data, WriteFlags::empty()).unwrap();
        }
        c_txn.commit().unwrap();
        master_txn.commit_and_renew().unwrap();

        let rtxn = master_txn.txn();

        let cursor = rtxn.open_ro_cursor(db).unwrap();
        let vals = cursor_dump(cursor);
        assert_eq!(vals.len(), items.len(), "must have duplicate records");
    }
}

1	use std::{fs, ops::Deref};
2
3	use crate::errors::CacheError;
4	use dozer_storage::{
5	lmdb::EnvironmentFlags,
6	lmdb_storage::{LmdbEnvironmentManager, LmdbEnvironmentOptions},
7	};
8	use tempdir::TempDir;
9
10	use super::cache::{CacheCommonOptions, CacheWriteOptions};
11
12	#[derive(Clone, Debug, Default)]	5✔
13	pub struct CacheOptions {
14	pub common: CacheCommonOptions,
15	pub kind: CacheOptionsKind,
16	}
17
18	#[derive(Clone, Debug, Default)]	×
19	pub struct CacheReadOptions {}
20
21	#[derive(Clone, Debug)]	×
22	pub enum CacheOptionsKind {
23	// Write Options
24	Write(CacheWriteOptions),
25
26	// Read Options
27	ReadOnly(CacheReadOptions),
28	}
29
30	impl Default for CacheOptionsKind {
31	fn default() -> Self {	5✔
32	Self::Write(CacheWriteOptions::default())	5✔
33	}	5✔
34	}
35
36	pub fn init_env(options: &CacheOptions) -> Result<(LmdbEnvironmentManager, String), CacheError> {	304✔
37	match &options.kind {	304✔
38	CacheOptionsKind::Write(write_options) => {	161✔
39	let (base_path, name, _temp_dir) = match &options.common.path {	161✔
40	None => {
41	let base_path = TempDir::new("dozer")?;	16✔
42	(	16✔
43	base_path.path().to_path_buf(),	16✔
44	"dozer-cache",	16✔
45	Some(base_path),	16✔
46	)	16✔
47	}
48	Some((base_path, name)) => {	145✔
49	fs::create_dir_all(base_path)?;	145✔
50	(base_path.clone(), name.deref(), None)	145✔
51	}
52	};
53
54	let options = LmdbEnvironmentOptions::new(	161✔
55	options.common.max_db_size,	161✔
56	options.common.max_readers,	161✔
57	write_options.max_size,	161✔
58	EnvironmentFlags::empty(),	161✔
59	);	161✔
60		161✔
61	Ok((	161✔
62	LmdbEnvironmentManager::create(&base_path, name, options)?,	161✔
63	name.to_string(),	161✔
64	))
65	}
66	CacheOptionsKind::ReadOnly(_) => {
67	let (base_path, name) = options	143✔
68	.common	143✔
69	.path	143✔
70	.as_ref()	143✔
71	.ok_or(CacheError::PathNotInitialized)?;	143✔
72
73	let env_options = LmdbEnvironmentOptions {	143✔
74	max_dbs: options.common.max_db_size,	143✔
75	max_readers: options.common.max_readers,	143✔
76	flags: EnvironmentFlags::READ_ONLY,	143✔
77	..Default::default()	143✔
78	};	143✔
79		143✔
80	Ok((	143✔
81	LmdbEnvironmentManager::create(base_path, name, env_options)?,	143✔
82	name.to_string(),	143✔
83	))
84	}
85	}
86	}	304✔
87
88	#[cfg(test)]
89	mod tests {
90	use dozer_storage::lmdb::{Cursor, DatabaseFlags, RoCursor, Transaction, WriteFlags};
91	use dozer_types::types::Field;
92
93	use crate::cache::lmdb::utils::{init_env, CacheOptions};
94
95	fn cursor_dump(mut cursor: RoCursor) -> Vec<(&[u8], &[u8])> {	1✔
96	cursor	1✔
97	.iter_dup()	1✔
98	.flatten()	1✔
99	.collect::<dozer_storage::lmdb::Result<Vec<_>>>()	1✔
100	.unwrap()	1✔
101	}	1✔
102
103	#[test]	1✔
104	fn duplicate_test_nested() {	1✔
105	let options = CacheOptions::default();	1✔
106	let mut env = init_env(&options).unwrap().0;	1✔
107		1✔
108	let db = env	1✔
109	.create_database(	1✔
110	Some("test"),	1✔
111	Some(DatabaseFlags::DUP_SORT \| DatabaseFlags::INTEGER_KEY),	1✔
112	)	1✔
113	.unwrap();	1✔
114		1✔
115	let txn = env.create_txn().unwrap();	1✔
116	let mut master_txn = txn.write();	1✔
117	let txn = master_txn.txn_mut();	1✔
118		1✔
119	let mut c_txn = txn.begin_nested_txn().unwrap();	1✔
120		1✔
121	let items: Vec<(i64, &[u8])> = vec![	1✔
122	(1, b"a"),	1✔
123	(2, b"a"),	1✔
124	(3, b"a"),	1✔
125	(1, b"b"),	1✔
126	(2, b"b"),	1✔
127	(3, b"b"),	1✔
128	(1, b"c"),	1✔
129	(2, b"c"),	1✔
130	(3, b"c"),	1✔
131	(1, b"e"),	1✔
132	(2, b"e"),	1✔
133	(3, b"e"),	1✔
134	];	1✔
135	for (key, data) in &items {	13✔
136	let key = [	12✔
137	"idx".as_bytes().to_vec(),	12✔
138	Field::Int(*key).encode(),	12✔
139	key.to_be_bytes().to_vec(),	12✔
140	]	12✔
141	.join("#".as_bytes());	12✔
142	c_txn.put(db, &key, data, WriteFlags::empty()).unwrap();	12✔
143	}	12✔
144	c_txn.commit().unwrap();	1✔
145	master_txn.commit_and_renew().unwrap();	1✔
146		1✔
147	let rtxn = master_txn.txn();	1✔
148		1✔
149	let cursor = rtxn.open_ro_cursor(db).unwrap();	1✔
150	let vals = cursor_dump(cursor);	1✔
151	assert_eq!(vals.len(), items.len(), "must have duplicate records");	1✔
152	}	1✔
153	}

getdozer / dozer / 4377467257

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