4113913291

Build Type

github

Committed by GitHub

Commit Message

Merge a8cca3f0b into 8f74ec17e

Pull Request Pull Request #821: refactor: Make `LmdbRoCache` and `LmdbRwCache` `Send` and `Sync`

Run Details

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

23486 of 37503 relevant lines covered (62.62%)

36806.72 hits per line

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

66.9

/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::{CacheOptions, CacheOptionsKind};

pub fn init_env(options: &CacheOptions) -> Result<LmdbEnvironmentManager, 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").map_err(|e| CacheError::Internal(Box::new(e)))?;
                    (
                        base_path.path().to_path_buf(),
                        "dozer-cache",
                        Some(base_path),
                    )
                }
                Some((base_path, name)) => {
                    fs::create_dir_all(base_path).map_err(|e| CacheError::Internal(Box::new(e)))?;
                    (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)?)
        }
        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,
            )?)
        }
    }
}

#[cfg(test)]
mod tests {
    use dozer_storage::{
        lmdb::{Cursor, DatabaseFlags, RoCursor, Transaction, WriteFlags},
        lmdb_storage::LmdbTransaction,
    };
    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();

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

getdozer / dozer / 4113913291

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