4309615408

Build Type

push

github

Committed by GitHub

Commit Message

chore: Remove an unused `Arc<RwLock>` (#1106)

Run Details

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

28466 of 40109 relevant lines covered (70.97%)

50957.01 hits per line

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

75.37

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

use crate::errors::{CacheError, IndexError};
use dozer_storage::lmdb::RwTransaction;
use dozer_types::types::{Field, IndexDefinition, Record, Schema};
use itertools::Itertools;
use unicode_segmentation::UnicodeSegmentation;

use crate::cache::index::{self, get_full_text_secondary_index};

use super::cache::SecondaryIndexDatabases;

pub struct Indexer<'a> {
    pub secondary_indexes: &'a SecondaryIndexDatabases,
}
impl<'a> Indexer<'a> {
    pub fn build_indexes(
        &self,
        txn: &mut RwTransaction,
        record: &Record,
        schema: &Schema,
        secondary_indexes: &[IndexDefinition],
        id: [u8; 8],
    ) -> Result<(), CacheError> {
        let schema_id = schema.identifier.ok_or(CacheError::SchemaHasNoIdentifier)?;

        if secondary_indexes.is_empty() {
            return Err(CacheError::Index(IndexError::MissingSecondaryIndexes));
        }
        for (idx, index) in secondary_indexes.iter().enumerate() {
            let db = *self
                .secondary_indexes
                .get(&(schema_id, idx))
                .ok_or(CacheError::SecondaryIndexDatabaseNotFound)?;

            match index {
                IndexDefinition::SortedInverted(fields) => {
                    let secondary_key = Self::_build_index_sorted_inverted(fields, &record.values);
                    db.insert(txn, &secondary_key, id)?;
                }
                IndexDefinition::FullText(field_index) => {
                    for secondary_key in
                        Self::_build_indices_full_text(*field_index, &record.values)?
                    {
                        db.insert(txn, &secondary_key, id)?;
                    }
                }
            }
        }
        Ok(())
    }

    pub fn delete_indexes(
        &self,
        txn: &mut RwTransaction,
        record: &Record,
        schema: &Schema,
        secondary_indexes: &[IndexDefinition],
        id: [u8; 8],
    ) -> Result<(), CacheError> {
        let schema_id = schema.identifier.ok_or(CacheError::SchemaHasNoIdentifier)?;
        for (idx, index) in secondary_indexes.iter().enumerate() {
            let db = *self
                .secondary_indexes
                .get(&(schema_id, idx))
                .ok_or(CacheError::SecondaryIndexDatabaseNotFound)?;

            match index {
                IndexDefinition::SortedInverted(fields) => {
                    let secondary_key = Self::_build_index_sorted_inverted(fields, &record.values);
                    db.delete(txn, &secondary_key, id)?;
                }
                IndexDefinition::FullText(field_index) => {
                    for secondary_key in
                        Self::_build_indices_full_text(*field_index, &record.values)?
                    {
                        db.delete(txn, &secondary_key, id)?;
                    }
                }
            }
        }

        Ok(())
    }

    fn _build_index_sorted_inverted(fields: &[usize], values: &[Field]) -> Vec<u8> {
        let values = fields
            .iter()
            .copied()
            .filter_map(|index| (values.get(index)))
            .collect::<Vec<_>>();
        // `values.len() == 1` criteria must be kept the same with `comparator.rs`.
        index::get_secondary_index(&values, values.len() == 1)
    }

    fn _build_indices_full_text(
        field_index: usize,
        values: &[Field],
    ) -> Result<Vec<Vec<u8>>, CacheError> {
        let Some(field) = values.get(field_index) else {
            return Err(CacheError::Index(IndexError::FieldIndexOutOfRange));
        };

        let string = match field {
            Field::String(string) => string,
            Field::Text(string) => string,
            Field::Null => "",
            _ => {
                return Err(CacheError::Index(IndexError::FieldNotCompatibleIndex(
                    field_index,
                )))
            }
        };

        Ok(string
            .unicode_words()
            .map(get_full_text_secondary_index)
            .unique()
            .collect())
    }
}

#[cfg(test)]
mod tests {
    use crate::cache::{
        lmdb::{
            cache::LmdbRwCache,
            tests::utils::{self as lmdb_utils, create_cache},
        },
        test_utils, RwCache,
    };

    use super::*;

    #[test]
    fn test_secondary_indexes() {
        let schema_name = "sample";
        let (cache, schema, secondary_indexes) = create_cache(schema_name, test_utils::schema_1);

        let items = vec![
            (1, Some("a".to_string()), Some(521)),
            (2, Some("a".to_string()), None),
            (3, None, Some(521)),
            (4, None, None),
        ];

        for val in items.clone() {
            lmdb_utils::insert_rec_1(&cache, &schema, val);
        }
        // No of index dbs
        let indexes = lmdb_utils::get_indexes(&cache);

        let index_count = indexes.iter().flatten().count();
        let expected_count = secondary_indexes.len();
        // 3 columns, 1 compound, 1 descending
        assert_eq!(
            indexes.len(),
            expected_count,
            "Must create db for each index"
        );

        assert_eq!(
            index_count,
            items.len() * expected_count,
            "Must index each field"
        );

        for a in [1i64, 2, 3, 4] {
            cache.delete(&Field::Int(a).encode()).unwrap();
        }

        assert_eq!(
            lmdb_utils::get_indexes(&cache)
                .into_iter()
                .flatten()
                .count(),
            0,
            "Must delete every index"
        );
    }

    #[test]
    fn test_build_indices_full_text() {
        let field_index = 0;
        assert_eq!(
            Indexer::_build_indices_full_text(
                field_index,
                &[Field::String("today is a good day".into())]
            )
            .unwrap(),
            vec![
                get_full_text_secondary_index("today"),
                get_full_text_secondary_index("is"),
                get_full_text_secondary_index("a"),
                get_full_text_secondary_index("good"),
                get_full_text_secondary_index("day"),
            ]
        );
    }

    #[test]
    fn test_full_text_secondary_index_with_duplicated_words() {
        let schema_name = "sample";
        let (schema, secondary_indexes) = test_utils::schema_full_text();
        let cache = LmdbRwCache::create(
            [(schema_name.to_string(), schema.clone(), secondary_indexes)],
            Default::default(),
            Default::default(),
        )
        .unwrap();

        let items = vec![(
            Some("another test".to_string()),
            Some("regular test regular".to_string()),
        )];

        for val in items {
            lmdb_utils::insert_full_text(&cache, &schema, val);
        }

        {
            let a = "another test".to_string();
            cache.delete(&Field::String(a).encode()).unwrap();
        }

        assert_eq!(
            lmdb_utils::get_indexes(&cache)
                .into_iter()
                .flatten()
                .count(),
            0,
            "Must delete every index"
        );
    }
}

1	use crate::errors::{CacheError, IndexError};
2	use dozer_storage::lmdb::RwTransaction;
3	use dozer_types::types::{Field, IndexDefinition, Record, Schema};
4	use itertools::Itertools;
5	use unicode_segmentation::UnicodeSegmentation;
6
7	use crate::cache::index::{self, get_full_text_secondary_index};
8
9	use super::cache::SecondaryIndexDatabases;
10
11	pub struct Indexer<'a> {
12	pub secondary_indexes: &'a SecondaryIndexDatabases,
13	}
14	impl<'a> Indexer<'a> {
15	pub fn build_indexes(	11,978✔
16	&self,	11,978✔
17	txn: &mut RwTransaction,	11,978✔
18	record: &Record,	11,978✔
19	schema: &Schema,	11,978✔
20	secondary_indexes: &[IndexDefinition],	11,978✔
21	id: [u8; 8],	11,978✔
22	) -> Result<(), CacheError> {	11,978✔
23	let schema_id = schema.identifier.ok_or(CacheError::SchemaHasNoIdentifier)?;	11,978✔
24		×
25	if secondary_indexes.is_empty() {	11,978✔
26	return Err(CacheError::Index(IndexError::MissingSecondaryIndexes));	×
27	}	11,978✔
28	for (idx, index) in secondary_indexes.iter().enumerate() {	83,780✔
29	let db = *self	83,780✔
30	.secondary_indexes	83,780✔
31	.get(&(schema_id, idx))	83,780✔
32	.ok_or(CacheError::SecondaryIndexDatabaseNotFound)?;	83,780✔
33		×
34	match index {	83,780✔
35	IndexDefinition::SortedInverted(fields) => {	77,763✔
36	let secondary_key = Self::_build_index_sorted_inverted(fields, &record.values);	77,763✔
37	db.insert(txn, &secondary_key, id)?;	77,763✔
38	}
39	IndexDefinition::FullText(field_index) => {	6,017✔
40	for secondary_key in	20,730✔
41	Self::_build_indices_full_text(*field_index, &record.values)?	6,017✔
42	{	×
43	db.insert(txn, &secondary_key, id)?;	20,730✔
44	}	×
45	}	×
46	}	×
47	}
48	Ok(())	11,978✔
49	}	11,978✔
50
51	pub fn delete_indexes(	13✔
52	&self,	13✔
53	txn: &mut RwTransaction,	13✔
54	record: &Record,	13✔
55	schema: &Schema,	13✔
56	secondary_indexes: &[IndexDefinition],	13✔
57	id: [u8; 8],	13✔
58	) -> Result<(), CacheError> {	13✔
59	let schema_id = schema.identifier.ok_or(CacheError::SchemaHasNoIdentifier)?;	13✔
60	for (idx, index) in secondary_indexes.iter().enumerate() {	32✔
61	let db = *self	32✔
62	.secondary_indexes	32✔
63	.get(&(schema_id, idx))	32✔
64	.ok_or(CacheError::SecondaryIndexDatabaseNotFound)?;	32✔
65		×
66	match index {	32✔
67	IndexDefinition::SortedInverted(fields) => {	30✔
68	let secondary_key = Self::_build_index_sorted_inverted(fields, &record.values);	30✔
69	db.delete(txn, &secondary_key, id)?;	30✔
70	}	×
71	IndexDefinition::FullText(field_index) => {	2✔
72	for secondary_key in	4✔
73	Self::_build_indices_full_text(*field_index, &record.values)?	2✔
74	{	×
75	db.delete(txn, &secondary_key, id)?;	4✔
76	}
77	}	×
78	}	×
79	}	×
80
81	Ok(())	13✔
82	}	13✔
83
84	fn _build_index_sorted_inverted(fields: &[usize], values: &[Field]) -> Vec<u8> {	77,793✔
85	let values = fields	77,793✔
86	.iter()	77,793✔
87	.copied()	77,793✔
88	.filter_map(\|index\| (values.get(index)))	107,814✔
89	.collect::<Vec<_>>();	77,793✔
90	// `values.len() == 1` criteria must be kept the same with `comparator.rs`.	77,793✔
91	index::get_secondary_index(&values, values.len() == 1)	77,793✔
92	}	77,793✔
93		×
94	fn _build_indices_full_text(	6,020✔
95	field_index: usize,	6,020✔
96	values: &[Field],	6,020✔
97	) -> Result<Vec<Vec<u8>>, CacheError> {	6,020✔
98	let Some(field) = values.get(field_index) else {	6,020✔
99	return Err(CacheError::Index(IndexError::FieldIndexOutOfRange));	×
100	};	×
101		×
102	let string = match field {	6,020✔
103	Field::String(string) => string,	6,017✔
104	Field::Text(string) => string,	3✔
105	Field::Null => "",	×
106	_ => {
107	return Err(CacheError::Index(IndexError::FieldNotCompatibleIndex(	×
108	field_index,	×
109	)))	×
110	}	×
111	};	×
112
113	Ok(string	6,020✔
114	.unicode_words()	6,020✔
115	.map(get_full_text_secondary_index)	6,020✔
116	.unique()	6,020✔
117	.collect())	6,020✔
118	}	6,020✔
119	}	×
120		×
121	#[cfg(test)]	×
122	mod tests {	×
123	use crate::cache::{	×
124	lmdb::{	×
125	cache::LmdbRwCache,
126	tests::utils::{self as lmdb_utils, create_cache},
127	},
128	test_utils, RwCache,
129	};
130
131	use super::*;
132
133	#[test]	1✔
134	fn test_secondary_indexes() {	1✔
135	let schema_name = "sample";	1✔
136	let (cache, schema, secondary_indexes) = create_cache(schema_name, test_utils::schema_1);	1✔
137		1✔
138	let items = vec![	1✔
139	(1, Some("a".to_string()), Some(521)),	1✔
140	(2, Some("a".to_string()), None),	1✔
141	(3, None, Some(521)),	1✔
142	(4, None, None),	1✔
143	];	1✔
144		×
145	for val in items.clone() {	4✔
146	lmdb_utils::insert_rec_1(&cache, &schema, val);	4✔
147	}	4✔
148	// No of index dbs	×
149	let indexes = lmdb_utils::get_indexes(&cache);	1✔
150		1✔
151	let index_count = indexes.iter().flatten().count();	1✔
152	let expected_count = secondary_indexes.len();	1✔
153	// 3 columns, 1 compound, 1 descending	1✔
154	assert_eq!(	1✔
155	indexes.len(),	1✔
156	expected_count,	×
157	"Must create db for each index"	×
158	);	×
159		×
160	assert_eq!(	1✔
161	index_count,	1✔
162	items.len() * expected_count,	1✔
163	"Must index each field"	×
164	);
165
166	for a in [1i64, 2, 3, 4] {	5✔
167	cache.delete(&Field::Int(a).encode()).unwrap();	4✔
168	}	4✔
169		×
170	assert_eq!(	1✔
171	lmdb_utils::get_indexes(&cache)	1✔
172	.into_iter()	1✔
173	.flatten()	1✔
174	.count(),	1✔
175	0,
176	"Must delete every index"	×
177	);	×
178	}	1✔
179		×
180	#[test]	1✔
181	fn test_build_indices_full_text() {	1✔
182	let field_index = 0;	1✔
183	assert_eq!(	1✔
184	Indexer::_build_indices_full_text(	1✔
185	field_index,	1✔
186	&[Field::String("today is a good day".into())]	1✔
187	)	1✔
188	.unwrap(),	1✔
189	vec![	1✔
190	get_full_text_secondary_index("today"),	1✔
191	get_full_text_secondary_index("is"),	1✔
192	get_full_text_secondary_index("a"),	1✔
193	get_full_text_secondary_index("good"),	1✔
194	get_full_text_secondary_index("day"),	1✔
195	]	1✔
196	);	1✔
197	}	1✔
198		×
199	#[test]	1✔
200	fn test_full_text_secondary_index_with_duplicated_words() {	1✔
201	let schema_name = "sample";	1✔
202	let (schema, secondary_indexes) = test_utils::schema_full_text();	1✔
203	let cache = LmdbRwCache::create(	1✔
204	[(schema_name.to_string(), schema.clone(), secondary_indexes)],	1✔
205	Default::default(),	1✔
206	Default::default(),	1✔
207	)	1✔
208	.unwrap();	1✔
209		1✔
210	let items = vec![(	1✔
211	Some("another test".to_string()),	1✔
212	Some("regular test regular".to_string()),	1✔
213	)];	1✔
214		×
215	for val in items {	2✔
216	lmdb_utils::insert_full_text(&cache, &schema, val);	1✔
217	}	1✔
218		×
219	{	1✔
220	let a = "another test".to_string();	1✔
221	cache.delete(&Field::String(a).encode()).unwrap();	1✔
222	}	1✔
223		1✔
224	assert_eq!(	1✔
225	lmdb_utils::get_indexes(&cache)	1✔
226	.into_iter()	1✔
227	.flatten()	1✔
228	.count(),	1✔
229	0,	×
230	"Must delete every index"	×
231	);	×
232	}	1✔
233	}	×

getdozer / dozer / 4309615408

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