4392484403

Build Type

push

github

Committed by GitHub

Commit Message

feat: Asynchoronous indexing (#1206)

Run Details

270 of 270 new or added lines in 13 files covered. (100.0%)

28714 of 38777 relevant lines covered (74.05%)

89484.24 hits per line

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

95.71

/dozer-cache/src/cache/lmdb/cache/query/tests.rs

use crate::cache::{
    expression::{FilterExpression, Operator, QueryExpression},
    lmdb::tests::utils::{create_cache, insert_rec_1},
    test_utils::{query_from_filter, schema_1, schema_full_text, schema_multi_indices},
    RecordWithId, RoCache, RwCache,
};
use dozer_types::{
    serde_json::{from_value, json, Value},
    types::{Field, Record, Schema},
};

#[test]
fn query_secondary_sorted_inverted() {
    let (cache, mut indexing_thread_pool, schema, _) = create_cache(schema_1);

    let mut record = Record::new(
        schema.identifier,
        vec![
            Field::Int(1),
            Field::String("test".to_string()),
            Field::Int(2),
        ],
        None,
    );

    cache.insert(&mut record).unwrap();
    assert!(record.version.is_some());
    cache.commit().unwrap();
    indexing_thread_pool.wait_until_catchup();

    let filter = FilterExpression::And(vec![
        FilterExpression::Simple("a".to_string(), Operator::EQ, Value::from(1)),
        FilterExpression::Simple(
            "b".to_string(),
            Operator::EQ,
            Value::from("test".to_string()),
        ),
    ]);

    // Query with an expression
    let query = query_from_filter(filter);

    let records = cache.query(&query).unwrap();
    assert_eq!(cache.count(&query).unwrap(), 1);
    assert_eq!(records.len(), 1, "must be equal");
    assert_eq!(records[0].record, record, "must be equal");
}

#[test]
fn query_secondary_full_text() {
    let (cache, mut indexing_thread_pool, schema, _) = create_cache(schema_full_text);

    let mut record = Record::new(
        schema.identifier,
        vec![
            Field::String("today is a good day".into()),
            Field::Text("marry has a little lamb".into()),
        ],
        None,
    );

    cache.insert(&mut record).unwrap();
    assert!(record.version.is_some());
    cache.commit().unwrap();
    indexing_thread_pool.wait_until_catchup();

    let filter = FilterExpression::Simple("foo".into(), Operator::Contains, "good".into());

    let query = query_from_filter(filter);

    let records = cache.query(&query).unwrap();
    assert_eq!(cache.count(&query).unwrap(), 1);
    assert_eq!(records.len(), 1);
    assert_eq!(records[0].record, record);

    let filter = FilterExpression::Simple("bar".into(), Operator::Contains, "lamb".into());
    let query = query_from_filter(filter);
    let records = cache.query(&query).unwrap();
    assert_eq!(cache.count(&query).unwrap(), 1);
    assert_eq!(records.len(), 1);
    assert_eq!(records[0].record, record);
}

#[test]
fn query_secondary_vars() {
    let (cache, mut indexing_thread_pool, schema, _) = create_cache(schema_1);

    let items = vec![
        (1, Some("yuri".to_string()), Some(521)),
        (2, Some("mega".to_string()), Some(521)),
        (3, Some("james".to_string()), Some(523)),
        (4, Some("james".to_string()), Some(524)),
        (5, Some("steff".to_string()), Some(526)),
        (6, Some("mega".to_string()), Some(527)),
        (7, Some("james".to_string()), Some(528)),
        (8, Some("ava".to_string()), None),
    ];
    // 26 alphabets
    for val in items {
        insert_rec_1(&cache, &schema, val);
    }
    cache.commit().unwrap();
    indexing_thread_pool.wait_until_catchup();

    test_query(json!({}), 8, &cache);

    test_query(
        json!({
            "$order_by": { "c": "desc" }
        }),
        8,
        &cache,
    );

    test_query(json!({"$filter":{ "a": {"$eq": 1}}}), 1, &cache);

    test_query(json!({"$filter":{ "a": {"$eq": null}}}), 0, &cache);

    test_query(json!({"$filter":{ "c": {"$eq": 521}}}), 2, &cache);

    test_query(json!({"$filter":{ "c": {"$eq": null}}}), 1, &cache);

    test_query(
        json!({"$filter":{ "a": 1, "b": "yuri".to_string()}}),
        1,
        &cache,
    );

    // No compound index for a,c
    test_query_err(json!({"$filter":{ "a": 1, "c": 521}}), &cache);

    test_query(
        json!({
            "$filter":{ "c": {"$eq": 521}},
            "$order_by": { "c": "asc" }
        }),
        2,
        &cache,
    );

    test_query_record(
        json!({
            "$filter":{ "a": {"$eq": 1}},
            "$order_by": { "b": "asc" }
        }),
        vec![(0, 1, "yuri".to_string(), 521)],
        &schema,
        &cache,
    );

    // Range tests
    test_query(json!({"$filter":{ "c": {"$lte": null}}}), 0, &cache);

    test_query(json!({"$filter":{ "c": {"$lte": 521}}}), 2, &cache);

    test_query(json!({"$filter":{ "c": {"$gte": 521}}}), 7, &cache);

    test_query(json!({"$filter":{ "c": {"$gt": 521}}}), 5, &cache);

    test_query(json!({"$filter":{ "c": {"$lte": 524}}}), 4, &cache);

    test_query(json!({"$filter":{ "c": {"$lt": 524}}}), 3, &cache);

    test_query(json!({"$filter":{ "c": {"$lt": 600}}}), 7, &cache);

    test_query(json!({"$filter":{ "c": {"$gt": 200}}}), 7, &cache);

    test_query_record(
        json!({
            "$filter":{ "c": {"$gt": 526}},
            "$order_by": { "c": "asc" }
        }),
        vec![
            (5, 6, "mega".to_string(), 527),
            (6, 7, "james".to_string(), 528),
        ],
        &schema,
        &cache,
    );

    test_query_record(
        json!({
            "$filter":{ "c": {"$gt": 526}},
            "$order_by": { "c": "desc" }
        }),
        vec![
            (6, 7, "james".to_string(), 528),
            (5, 6, "mega".to_string(), 527),
        ],
        &schema,
        &cache,
    );
}

#[test]
fn query_secondary_multi_indices() {
    let (cache, mut indexing_thread_pool, schema, _) = create_cache(schema_multi_indices);

    for (id, text) in [
        (1, "apple ball cake dance"),
        (2, "ball cake dance egg"),
        (3, "cake dance egg fish"),
        (4, "dance egg fish glove"),
        (5, "egg fish glove heart"),
        (6, "fish glove heart igloo"),
        (7, "glove heart igloo jump"),
    ] {
        let mut record = Record {
            schema_id: schema.identifier,
            values: vec![Field::Int(id), Field::String(text.into())],
            version: None,
        };
        cache.insert(&mut record).unwrap();
        assert!(record.version.is_some());
    }
    cache.commit().unwrap();
    indexing_thread_pool.wait_until_catchup();

    let query = query_from_filter(FilterExpression::And(vec![
        FilterExpression::Simple("id".into(), Operator::GT, Value::from(2)),
        FilterExpression::Simple("text".into(), Operator::Contains, Value::from("dance")),
    ]));

    let records = cache.query(&query).unwrap();
    assert_eq!(cache.count(&query).unwrap(), 2);
    assert_eq!(
        records,
        vec![
            RecordWithId::new(
                2,
                Record {
                    schema_id: schema.identifier,
                    values: vec![Field::Int(3), Field::String("cake dance egg fish".into())],
                    version: Some(1)
                }
            ),
            RecordWithId::new(
                3,
                Record {
                    schema_id: schema.identifier,
                    values: vec![Field::Int(4), Field::String("dance egg fish glove".into())],
                    version: Some(1)
                }
            ),
        ]
    );
}

fn test_query_err(query: Value, cache: &dyn RwCache) {
    let query = from_value::<QueryExpression>(query).unwrap();
    let count_result = cache.count(&query);
    let result = cache.query(&query);

    assert!(matches!(
        count_result.unwrap_err(),
        crate::errors::CacheError::Plan(_)
    ),);
    assert!(matches!(
        result.unwrap_err(),
        crate::errors::CacheError::Plan(_)
    ),);
}
fn test_query(query: Value, count: usize, cache: &dyn RwCache) {
    let query = from_value::<QueryExpression>(query).unwrap();
    assert_eq!(cache.count(&query).unwrap(), count);
    let records = cache.query(&query).unwrap();

    assert_eq!(records.len(), count, "Count must be equal : {query:?}");
}

fn test_query_record(
    query: Value,
    expected: Vec<(u64, i64, String, i64)>,
    schema: &Schema,
    cache: &dyn RwCache,
) {
    let query = from_value::<QueryExpression>(query).unwrap();
    assert_eq!(cache.count(&query).unwrap(), expected.len());
    let records = cache.query(&query).unwrap();
    let expected = expected
        .into_iter()
        .map(|(id, a, b, c)| {
            RecordWithId::new(
                id,
                Record::new(
                    schema.identifier,
                    vec![Field::Int(a), Field::String(b), Field::Int(c)],
                    Some(1),
                ),
            )
        })
        .collect::<Vec<_>>();
    assert_eq!(records, expected);
}

1	use crate::cache::{
2	expression::{FilterExpression, Operator, QueryExpression},
3	lmdb::tests::utils::{create_cache, insert_rec_1},
4	test_utils::{query_from_filter, schema_1, schema_full_text, schema_multi_indices},
5	RecordWithId, RoCache, RwCache,
6	};
7	use dozer_types::{
8	serde_json::{from_value, json, Value},
9	types::{Field, Record, Schema},
10	};
11
12	#[test]	1✔
13	fn query_secondary_sorted_inverted() {	1✔
14	let (cache, mut indexing_thread_pool, schema, _) = create_cache(schema_1);	1✔
15		1✔
16	let mut record = Record::new(	1✔
17	schema.identifier,	1✔
18	vec![	1✔
19	Field::Int(1),	1✔
20	Field::String("test".to_string()),	1✔
21	Field::Int(2),	1✔
22	],	1✔
23	None,	1✔
24	);	1✔
25		1✔
26	cache.insert(&mut record).unwrap();	1✔
27	assert!(record.version.is_some());	1✔
28	cache.commit().unwrap();	1✔
29	indexing_thread_pool.wait_until_catchup();	1✔
30		1✔
31	let filter = FilterExpression::And(vec![	1✔
32	FilterExpression::Simple("a".to_string(), Operator::EQ, Value::from(1)),	1✔
33	FilterExpression::Simple(	1✔
34	"b".to_string(),	1✔
35	Operator::EQ,	1✔
36	Value::from("test".to_string()),	1✔
37	),	1✔
38	]);	1✔
39		1✔
40	// Query with an expression	1✔
41	let query = query_from_filter(filter);	1✔
42		1✔
43	let records = cache.query(&query).unwrap();	1✔
44	assert_eq!(cache.count(&query).unwrap(), 1);	1✔
45	assert_eq!(records.len(), 1, "must be equal");	1✔
46	assert_eq!(records[0].record, record, "must be equal");	1✔
47	}	1✔
48		×
49	#[test]	1✔
50	fn query_secondary_full_text() {	1✔
51	let (cache, mut indexing_thread_pool, schema, _) = create_cache(schema_full_text);	1✔
52		1✔
53	let mut record = Record::new(	1✔
54	schema.identifier,	1✔
55	vec![	1✔
56	Field::String("today is a good day".into()),	1✔
57	Field::Text("marry has a little lamb".into()),	1✔
58	],	1✔
59	None,	1✔
60	);	1✔
61		1✔
62	cache.insert(&mut record).unwrap();	1✔
63	assert!(record.version.is_some());	1✔
64	cache.commit().unwrap();	1✔
65	indexing_thread_pool.wait_until_catchup();	1✔
66		1✔
67	let filter = FilterExpression::Simple("foo".into(), Operator::Contains, "good".into());	1✔
68		1✔
69	let query = query_from_filter(filter);	1✔
70		1✔
71	let records = cache.query(&query).unwrap();	1✔
72	assert_eq!(cache.count(&query).unwrap(), 1);	1✔
73	assert_eq!(records.len(), 1);	1✔
74	assert_eq!(records[0].record, record);	1✔
75		×
76	let filter = FilterExpression::Simple("bar".into(), Operator::Contains, "lamb".into());	1✔
77	let query = query_from_filter(filter);	1✔
78	let records = cache.query(&query).unwrap();	1✔
79	assert_eq!(cache.count(&query).unwrap(), 1);	1✔
80	assert_eq!(records.len(), 1);	1✔
81	assert_eq!(records[0].record, record);	1✔
82	}	1✔
83		×
84	#[test]	1✔
85	fn query_secondary_vars() {	1✔
86	let (cache, mut indexing_thread_pool, schema, _) = create_cache(schema_1);	1✔
87		1✔
88	let items = vec![	1✔
89	(1, Some("yuri".to_string()), Some(521)),	1✔
90	(2, Some("mega".to_string()), Some(521)),	1✔
91	(3, Some("james".to_string()), Some(523)),	1✔
92	(4, Some("james".to_string()), Some(524)),	1✔
93	(5, Some("steff".to_string()), Some(526)),	1✔
94	(6, Some("mega".to_string()), Some(527)),	1✔
95	(7, Some("james".to_string()), Some(528)),	1✔
96	(8, Some("ava".to_string()), None),	1✔
97	];	1✔
98	// 26 alphabets
99	for val in items {	9✔
100	insert_rec_1(&cache, &schema, val);	8✔
101	}	8✔
102	cache.commit().unwrap();	1✔
103	indexing_thread_pool.wait_until_catchup();	1✔
104		1✔
105	test_query(json!({}), 8, &cache);	1✔
106		1✔
107	test_query(	1✔
108	json!({	1✔
109	"$order_by": { "c": "desc" }	1✔
110	}),	1✔
111	8,	1✔
112	&cache,	1✔
113	);	1✔
114		1✔
115	test_query(json!({"$filter":{ "a": {"$eq": 1}}}), 1, &cache);	1✔
116		1✔
117	test_query(json!({"$filter":{ "a": {"$eq": null}}}), 0, &cache);	1✔
118		1✔
119	test_query(json!({"$filter":{ "c": {"$eq": 521}}}), 2, &cache);	1✔
120		1✔
121	test_query(json!({"$filter":{ "c": {"$eq": null}}}), 1, &cache);	1✔
122		1✔
123	test_query(	1✔
124	json!({"$filter":{ "a": 1, "b": "yuri".to_string()}}),	1✔
125	1,	1✔
126	&cache,	1✔
127	);	1✔
128		1✔
129	// No compound index for a,c	1✔
130	test_query_err(json!({"$filter":{ "a": 1, "c": 521}}), &cache);	1✔
131		1✔
132	test_query(	1✔
133	json!({	1✔
134	"$filter":{ "c": {"$eq": 521}},	1✔
135	"$order_by": { "c": "asc" }	1✔
136	}),	1✔
137	2,	1✔
138	&cache,	1✔
139	);	1✔
140		1✔
141	test_query_record(	1✔
142	json!({	1✔
143	"$filter":{ "a": {"$eq": 1}},	1✔
144	"$order_by": { "b": "asc" }	1✔
145	}),	1✔
146	vec![(0, 1, "yuri".to_string(), 521)],	1✔
147	&schema,	1✔
148	&cache,	1✔
149	);	1✔
150		1✔
151	// Range tests	1✔
152	test_query(json!({"$filter":{ "c": {"$lte": null}}}), 0, &cache);	1✔
153		1✔
154	test_query(json!({"$filter":{ "c": {"$lte": 521}}}), 2, &cache);	1✔
155		1✔
156	test_query(json!({"$filter":{ "c": {"$gte": 521}}}), 7, &cache);	1✔
157		1✔
158	test_query(json!({"$filter":{ "c": {"$gt": 521}}}), 5, &cache);	1✔
159		1✔
160	test_query(json!({"$filter":{ "c": {"$lte": 524}}}), 4, &cache);	1✔
161		1✔
162	test_query(json!({"$filter":{ "c": {"$lt": 524}}}), 3, &cache);	1✔
163		1✔
164	test_query(json!({"$filter":{ "c": {"$lt": 600}}}), 7, &cache);	1✔
165		1✔
166	test_query(json!({"$filter":{ "c": {"$gt": 200}}}), 7, &cache);	1✔
167		1✔
168	test_query_record(	1✔
169	json!({	1✔
170	"$filter":{ "c": {"$gt": 526}},	1✔
171	"$order_by": { "c": "asc" }	1✔
172	}),	1✔
173	vec![	1✔
174	(5, 6, "mega".to_string(), 527),	1✔
175	(6, 7, "james".to_string(), 528),	1✔
176	],	1✔
177	&schema,	1✔
178	&cache,	1✔
179	);	1✔
180		1✔
181	test_query_record(	1✔
182	json!({	1✔
183	"$filter":{ "c": {"$gt": 526}},	1✔
184	"$order_by": { "c": "desc" }	1✔
185	}),	1✔
186	vec![	1✔
187	(6, 7, "james".to_string(), 528),	1✔
188	(5, 6, "mega".to_string(), 527),	1✔
189	],	1✔
190	&schema,	1✔
191	&cache,	1✔
192	);	1✔
193	}	1✔
194		×
195	#[test]	1✔
196	fn query_secondary_multi_indices() {	1✔
197	let (cache, mut indexing_thread_pool, schema, _) = create_cache(schema_multi_indices);	1✔
198		×
199	for (id, text) in [	7✔
200	(1, "apple ball cake dance"),	1✔
201	(2, "ball cake dance egg"),	1✔
202	(3, "cake dance egg fish"),	1✔
203	(4, "dance egg fish glove"),	1✔
204	(5, "egg fish glove heart"),	1✔
205	(6, "fish glove heart igloo"),	1✔
206	(7, "glove heart igloo jump"),	1✔
207	] {	×
208	let mut record = Record {	7✔
209	schema_id: schema.identifier,	7✔
210	values: vec![Field::Int(id), Field::String(text.into())],	7✔
211	version: None,	7✔
212	};	7✔
213	cache.insert(&mut record).unwrap();	7✔
214	assert!(record.version.is_some());	7✔
215	}	×
216	cache.commit().unwrap();	1✔
217	indexing_thread_pool.wait_until_catchup();	1✔
218		1✔
219	let query = query_from_filter(FilterExpression::And(vec![	1✔
220	FilterExpression::Simple("id".into(), Operator::GT, Value::from(2)),	1✔
221	FilterExpression::Simple("text".into(), Operator::Contains, Value::from("dance")),	1✔
222	]));	1✔
223		1✔
224	let records = cache.query(&query).unwrap();	1✔
225	assert_eq!(cache.count(&query).unwrap(), 2);	1✔
226	assert_eq!(	1✔
227	records,	1✔
228	vec![	1✔
229	RecordWithId::new(	1✔
230	2,	1✔
231	Record {	1✔
232	schema_id: schema.identifier,	1✔
233	values: vec![Field::Int(3), Field::String("cake dance egg fish".into())],	1✔
234	version: Some(1)	1✔
235	}	1✔
236	),	1✔
237	RecordWithId::new(	1✔
238	3,	1✔
239	Record {	1✔
240	schema_id: schema.identifier,	1✔
241	values: vec![Field::Int(4), Field::String("dance egg fish glove".into())],	1✔
242	version: Some(1)	1✔
243	}	1✔
244	),	1✔
245	]	1✔
246	);	1✔
247	}	1✔
248
249	fn test_query_err(query: Value, cache: &dyn RwCache) {	1✔
250	let query = from_value::<QueryExpression>(query).unwrap();	1✔
251	let count_result = cache.count(&query);	1✔
252	let result = cache.query(&query);	1✔
253
254	assert!(matches!(	1✔
255	count_result.unwrap_err(),	1✔
256	crate::errors::CacheError::Plan(_)	×
257	),);	×
258	assert!(matches!(	1✔
259	result.unwrap_err(),	1✔
260	crate::errors::CacheError::Plan(_)	×
261	),);	×
262	}	1✔
263	fn test_query(query: Value, count: usize, cache: &dyn RwCache) {	16✔
264	let query = from_value::<QueryExpression>(query).unwrap();	16✔
265	assert_eq!(cache.count(&query).unwrap(), count);	16✔
266	let records = cache.query(&query).unwrap();	16✔
267		16✔
268	assert_eq!(records.len(), count, "Count must be equal : {query:?}");	16✔
269	}	16✔
270		×
271	fn test_query_record(	3✔
272	query: Value,	3✔
273	expected: Vec<(u64, i64, String, i64)>,	3✔
274	schema: &Schema,	3✔
275	cache: &dyn RwCache,	3✔
276	) {	3✔
277	let query = from_value::<QueryExpression>(query).unwrap();	3✔
278	assert_eq!(cache.count(&query).unwrap(), expected.len());	3✔
279	let records = cache.query(&query).unwrap();	3✔
280	let expected = expected	3✔
281	.into_iter()	3✔
282	.map(\|(id, a, b, c)\| {	5✔
283	RecordWithId::new(	5✔
284	id,	5✔
285	Record::new(	5✔
286	schema.identifier,	5✔
287	vec![Field::Int(a), Field::String(b), Field::Int(c)],	5✔
288	Some(1),	5✔
289	),	5✔
290	)	5✔
291	})	5✔
292	.collect::<Vec<_>>();	3✔
293	assert_eq!(records, expected);	3✔
294	}	3✔

getdozer / dozer / 4392484403

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