4382580286

Build Type

push

github

Committed by GitHub

Commit Message

feat: Separate cache operation log environment and index environments (#1199)

Run Details

1370 of 1370 new or added lines in 33 files covered. (100.0%)

28671 of 41023 relevant lines covered (69.89%)

51121.29 hits per line

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

79.55

/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};

pub fn init_env(
    common_options: &CacheCommonOptions,
    write_options: Option<CacheWriteOptions>,
) -> Result<(LmdbEnvironmentManager, String), CacheError> {
    if let Some(write_options) = write_options {
        create_env(common_options, write_options)
    } else {
        let (env, name) = open_env(common_options)?;
        Ok((env, name.to_string()))
    }
}

fn create_env(
    common_options: &CacheCommonOptions,
    write_options: CacheWriteOptions,
) -> Result<(LmdbEnvironmentManager, String), CacheError> {
    let (base_path, name, _temp_dir) = match &common_options.path {
        None => {
            let base_path =
                TempDir::new("dozer").map_err(|e| CacheError::Io("tempdir".into(), 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::Io(base_path.clone(), e))?;
            (base_path.clone(), name.deref(), None)
        }
    };

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

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

fn open_env(options: &CacheCommonOptions) -> Result<(LmdbEnvironmentManager, &str), CacheError> {
    let (base_path, name) = options
        .path
        .as_ref()
        .ok_or(CacheError::PathNotInitialized)?;

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

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

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

    use super::*;

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

getdozer / dozer / 4382580286

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