4698406303

Build Type

github

Committed by GitHub

Commit Message

Merge daefffe87 into b6889464a

Pull Request Pull Request #1426: feat: implement python log bindings

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

34863 of 45840 relevant lines covered (76.05%)

10764.36 hits per line

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

85.86

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

use std::{cmp::Ordering, ops::Bound};

use dozer_storage::lmdb::Transaction;
use dozer_types::{
    borrow::{Borrow, IntoOwned},
    types::{Field, IndexDefinition},
};

use crate::{
    cache::{
        expression::{Operator, SortDirection},
        index,
        lmdb::cache::secondary_environment::SecondaryEnvironment,
        plan::{IndexScanKind, SortedInvertedRangeQuery},
    },
    errors::{CacheError, IndexError},
};

use super::lmdb_cmp::lmdb_cmp;

pub fn build_index_scan<'txn, T: Transaction, S: SecondaryEnvironment>(
    secondary_txn: &'txn T,
    secondary_env: &S,
    index_scan_kind: &IndexScanKind,
) -> Result<impl Iterator<Item = Result<u64, CacheError>> + 'txn, CacheError> {
    let is_single_field_sorted_inverted =
        is_single_field_sorted_inverted(secondary_env.index_definition());
    let range = get_range_spec(index_scan_kind, is_single_field_sorted_inverted)?;

    let start = match &range.start {
        Some(KeyEndpoint::Including(key)) => Bound::Included(key.as_slice()),
        Some(KeyEndpoint::Excluding(key)) => Bound::Excluded(key.as_slice()),
        None => Bound::Unbounded,
    };

    let database = secondary_env.database().database();
    Ok(secondary_env
        .database()
        .range(
            secondary_txn,
            start,
            range.direction == SortDirection::Ascending,
        )?
        .take_while(move |result| match result {
            Ok((key, _)) => {
                if let Some(end_key) = &range.end {
                    match lmdb_cmp(secondary_txn, database, key.borrow(), end_key.key()) {
                        Ordering::Less => {
                            matches!(range.direction, SortDirection::Ascending)
                        }
                        Ordering::Equal => matches!(end_key, KeyEndpoint::Including(_)),
                        Ordering::Greater => {
                            matches!(range.direction, SortDirection::Descending)
                        }
                    }
                } else {
                    true
                }
            }
            Err(_) => true,
        })
        .map(|result| {
            result
                .map(|(_, id)| id.into_owned())
                .map_err(CacheError::Storage)
        }))
}

fn is_single_field_sorted_inverted(index: &IndexDefinition) -> bool {
    match index {
        // `fields.len() == 1` criteria must be kept the same with `comparator.rs`.
        IndexDefinition::SortedInverted(fields) => fields.len() == 1,
        _ => false,
    }
}

#[derive(Debug, Clone)]
pub enum KeyEndpoint {
    Including(Vec<u8>),
    Excluding(Vec<u8>),
}

impl KeyEndpoint {
    pub fn key(&self) -> &[u8] {
        match self {
            KeyEndpoint::Including(key) => key,
            KeyEndpoint::Excluding(key) => key,
        }
    }
}

#[derive(Debug)]
struct RangeSpec {
    start: Option<KeyEndpoint>,
    end: Option<KeyEndpoint>,
    direction: SortDirection,
}

fn get_range_spec(
    index_scan_kind: &IndexScanKind,
    is_single_field_sorted_inverted: bool,
) -> Result<RangeSpec, CacheError> {
    match &index_scan_kind {
        IndexScanKind::SortedInverted {
            eq_filters,
            range_query,
        } => {
            let comparison_key = build_sorted_inverted_comparison_key(
                eq_filters,
                range_query.as_ref(),
                is_single_field_sorted_inverted,
            );
            // There're 3 cases:
            // 1. Range query with operator.
            // 2. Range query without operator (only order by).
            // 3. No range query.
            Ok(if let Some(range_query) = range_query {
                match range_query.operator_and_value {
                    Some((operator, _)) => {
                        // Here we respond to case 1, examples are `a = 1 && b > 2` or `b < 2`.
                        let comparison_key = comparison_key.expect("here's at least a range query");
                        let null_key = build_sorted_inverted_comparison_key(
                            eq_filters,
                            Some(&SortedInvertedRangeQuery {
                                field_index: range_query.field_index,
                                operator_and_value: Some((operator, Field::Null)),
                                sort_direction: range_query.sort_direction,
                            }),
                            is_single_field_sorted_inverted,
                        )
                        .expect("we provided a range query");
                        get_key_interval_from_range_query(
                            comparison_key,
                            null_key,
                            operator,
                            range_query.sort_direction,
                        )
                    }
                    None => {
                        // Here we respond to case 2, examples are `a = 1 && b asc` or `b desc`.
                        if let Some(comparison_key) = comparison_key {
                            // This is the case like `a = 1 && b asc`. The comparison key is only built from `a = 1`.
                            // We use `a = 1 && b = null` as a sentinel, using the invariant that `null` is greater than anything.
                            let null_key = build_sorted_inverted_comparison_key(
                                eq_filters,
                                Some(&SortedInvertedRangeQuery {
                                    field_index: range_query.field_index,
                                    operator_and_value: Some((Operator::LT, Field::Null)),
                                    sort_direction: range_query.sort_direction,
                                }),
                                is_single_field_sorted_inverted,
                            )
                            .expect("we provided a range query");
                            match range_query.sort_direction {
                                SortDirection::Ascending => RangeSpec {
                                    start: Some(KeyEndpoint::Excluding(comparison_key)),
                                    end: Some(KeyEndpoint::Including(null_key)),
                                    direction: SortDirection::Ascending,
                                },
                                SortDirection::Descending => RangeSpec {
                                    start: Some(KeyEndpoint::Including(null_key)),
                                    end: Some(KeyEndpoint::Excluding(comparison_key)),
                                    direction: SortDirection::Descending,
                                },
                            }
                        } else {
                            // Just all of them.
                            RangeSpec {
                                start: None,
                                end: None,
                                direction: range_query.sort_direction,
                            }
                        }
                    }
                }
            } else {
                // Here we respond to case 3, examples are `a = 1` or `a = 1 && b = 2`.
                let comparison_key = comparison_key
                    .expect("here's at least a eq filter because there's no range query");
                RangeSpec {
                    start: Some(KeyEndpoint::Including(comparison_key.clone())),
                    end: Some(KeyEndpoint::Including(comparison_key)),
                    direction: SortDirection::Ascending, // doesn't matter
                }
            })
        }
        IndexScanKind::FullText { filter } => match filter.op {
            Operator::Contains => {
                let token = match &filter.val {
                    Field::String(token) => token,
                    Field::Text(token) => token,
                    _ => return Err(CacheError::Index(IndexError::ExpectedStringFullText)),
                };
                let key = index::get_full_text_secondary_index(token);
                Ok(RangeSpec {
                    start: Some(KeyEndpoint::Including(key.clone())),
                    end: Some(KeyEndpoint::Including(key)),
                    direction: SortDirection::Ascending, // doesn't matter
                })
            }
            Operator::MatchesAll | Operator::MatchesAny => {
                unimplemented!("matches all and matches any are not implemented")
            }
            other => panic!("operator {other:?} is not supported by full text index"),
        },
    }
}

fn build_sorted_inverted_comparison_key(
    eq_filters: &[(usize, Field)],
    range_query: Option<&SortedInvertedRangeQuery>,
    is_single_field_index: bool,
) -> Option<Vec<u8>> {
    let mut fields = vec![];
    eq_filters.iter().for_each(|filter| {
        fields.push(&filter.1);
    });
    if let Some(range_query) = range_query {
        if let Some((_, val)) = &range_query.operator_and_value {
            fields.push(val);
        }
    }
    if fields.is_empty() {
        None
    } else {
        Some(index::get_secondary_index(&fields, is_single_field_index))
    }
}

/// Here we use the invariant that `null` is greater than anything.
fn get_key_interval_from_range_query(
    comparison_key: Vec<u8>,
    null_key: Vec<u8>,
    operator: Operator,
    sort_direction: SortDirection,
) -> RangeSpec {
    match (operator, sort_direction) {
        (Operator::LT, SortDirection::Ascending) => RangeSpec {
            start: None,
            end: Some(KeyEndpoint::Excluding(comparison_key)),
            direction: SortDirection::Ascending,
        },
        (Operator::LT, SortDirection::Descending) => RangeSpec {
            start: Some(KeyEndpoint::Excluding(comparison_key)),
            end: None,
            direction: SortDirection::Descending,
        },
        (Operator::LTE, SortDirection::Ascending) => RangeSpec {
            start: None,
            end: Some(KeyEndpoint::Including(comparison_key)),
            direction: SortDirection::Ascending,
        },
        (Operator::LTE, SortDirection::Descending) => RangeSpec {
            start: Some(KeyEndpoint::Including(comparison_key)),
            end: None,
            direction: SortDirection::Descending,
        },
        (Operator::GT, SortDirection::Ascending) => RangeSpec {
            start: Some(KeyEndpoint::Excluding(comparison_key)),
            end: Some(KeyEndpoint::Excluding(null_key)),
            direction: SortDirection::Ascending,
        },
        (Operator::GT, SortDirection::Descending) => RangeSpec {
            start: Some(KeyEndpoint::Excluding(null_key)),
            end: Some(KeyEndpoint::Excluding(comparison_key)),
            direction: SortDirection::Descending,
        },
        (Operator::GTE, SortDirection::Ascending) => RangeSpec {
            start: Some(KeyEndpoint::Including(comparison_key)),
            end: Some(KeyEndpoint::Excluding(null_key)),
            direction: SortDirection::Ascending,
        },
        (Operator::GTE, SortDirection::Descending) => RangeSpec {
            start: Some(KeyEndpoint::Excluding(null_key)),
            end: Some(KeyEndpoint::Including(comparison_key)),
            direction: SortDirection::Descending,
        },
        (other, _) => {
            panic!("operator {other:?} is not supported by sorted inverted index range query")
        }
    }
}

1	use std::{cmp::Ordering, ops::Bound};
2
3	use dozer_storage::lmdb::Transaction;
4	use dozer_types::{
5	borrow::{Borrow, IntoOwned},
6	types::{Field, IndexDefinition},
7	};
8
9	use crate::{
10	cache::{
11	expression::{Operator, SortDirection},
12	index,
13	lmdb::cache::secondary_environment::SecondaryEnvironment,
14	plan::{IndexScanKind, SortedInvertedRangeQuery},
15	},
16	errors::{CacheError, IndexError},
17	};
18
19	use super::lmdb_cmp::lmdb_cmp;
20
21	pub fn build_index_scan<'txn, T: Transaction, S: SecondaryEnvironment>(	119✔
22	secondary_txn: &'txn T,	119✔
23	secondary_env: &S,	119✔
24	index_scan_kind: &IndexScanKind,	119✔
25	) -> Result<impl Iterator<Item = Result<u64, CacheError>> + 'txn, CacheError> {	119✔
26	let is_single_field_sorted_inverted =	119✔
27	is_single_field_sorted_inverted(secondary_env.index_definition());	119✔
28	let range = get_range_spec(index_scan_kind, is_single_field_sorted_inverted)?;	119✔
29
30	let start = match &range.start {	119✔
31	Some(KeyEndpoint::Including(key)) => Bound::Included(key.as_slice()),	97✔
32	Some(KeyEndpoint::Excluding(key)) => Bound::Excluded(key.as_slice()),	12✔
33	None => Bound::Unbounded,	10✔
34	};
35
36	let database = secondary_env.database().database();	119✔
37	Ok(secondary_env	119✔
38	.database()	119✔
39	.range(	119✔
40	secondary_txn,	119✔
41	start,	119✔
42	range.direction == SortDirection::Ascending,	119✔
43	)?	119✔
44	.take_while(move \|result\| match result {	1,456✔
45	Ok((key, _)) => {	1,456✔
46	if let Some(end_key) = &range.end {	1,456✔
47	match lmdb_cmp(secondary_txn, database, key.borrow(), end_key.key()) {	1,440✔
48	Ordering::Less => {
49	matches!(range.direction, SortDirection::Ascending)	80✔
50	}
51	Ordering::Equal => matches!(end_key, KeyEndpoint::Including(_)),	1,321✔
52	Ordering::Greater => {
53	matches!(range.direction, SortDirection::Descending)	39✔
54	}
55	}
56	} else {
57	true	16✔
58	}
59	}
60	Err(_) => true,	×
61	})	1,456✔
62	.map(\|result\| {	1,409✔
63	result	1,409✔
64	.map(\|(_, id)\| id.into_owned())	1,409✔
65	.map_err(CacheError::Storage)	1,409✔
66	}))	1,409✔
67	}	119✔
68
69	fn is_single_field_sorted_inverted(index: &IndexDefinition) -> bool {	119✔
70	match index {	119✔
71	// `fields.len() == 1` criteria must be kept the same with `comparator.rs`.
72	IndexDefinition::SortedInverted(fields) => fields.len() == 1,	113✔
73	_ => false,	6✔
74	}
75	}	119✔
76
77	#[derive(Debug, Clone)]	×
78	pub enum KeyEndpoint {
79	Including(Vec<u8>),
80	Excluding(Vec<u8>),
81	}
82
83	impl KeyEndpoint {
84	pub fn key(&self) -> &[u8] {	1,440✔
85	match self {	1,440✔
86	KeyEndpoint::Including(key) => key,	1,350✔
87	KeyEndpoint::Excluding(key) => key,	90✔
88	}
89	}	1,440✔
90	}
91
92	#[derive(Debug)]	×
93	struct RangeSpec {
94	start: Option<KeyEndpoint>,
95	end: Option<KeyEndpoint>,
96	direction: SortDirection,
97	}
98
99	fn get_range_spec(	119✔
100	index_scan_kind: &IndexScanKind,	119✔
101	is_single_field_sorted_inverted: bool,	119✔
102	) -> Result<RangeSpec, CacheError> {	119✔
103	match &index_scan_kind {	119✔
104	IndexScanKind::SortedInverted {
105	eq_filters,	113✔
106	range_query,	113✔
107	} => {	113✔
108	let comparison_key = build_sorted_inverted_comparison_key(	113✔
109	eq_filters,	113✔
110	range_query.as_ref(),	113✔
111	is_single_field_sorted_inverted,	113✔
112	);	113✔
113	// There're 3 cases:
114	// 1. Range query with operator.
115	// 2. Range query without operator (only order by).
116	// 3. No range query.
117	Ok(if let Some(range_query) = range_query {	113✔
118	match range_query.operator_and_value {	24✔
119	Some((operator, _)) => {	20✔
120	// Here we respond to case 1, examples are `a = 1 && b > 2` or `b < 2`.	20✔
121	let comparison_key = comparison_key.expect("here's at least a range query");	20✔
122	let null_key = build_sorted_inverted_comparison_key(	20✔
123	eq_filters,	20✔
124	Some(&SortedInvertedRangeQuery {	20✔
125	field_index: range_query.field_index,	20✔
126	operator_and_value: Some((operator, Field::Null)),	20✔
127	sort_direction: range_query.sort_direction,	20✔
128	}),	20✔
129	is_single_field_sorted_inverted,	20✔
130	)	20✔
131	.expect("we provided a range query");	20✔
132	get_key_interval_from_range_query(	20✔
133	comparison_key,	20✔
134	null_key,	20✔
135	operator,	20✔
136	range_query.sort_direction,	20✔
137	)	20✔
138	}
139	None => {
140	// Here we respond to case 2, examples are `a = 1 && b asc` or `b desc`.
141	if let Some(comparison_key) = comparison_key {	4✔
142	// This is the case like `a = 1 && b asc`. The comparison key is only built from `a = 1`.
143	// We use `a = 1 && b = null` as a sentinel, using the invariant that `null` is greater than anything.
144	let null_key = build_sorted_inverted_comparison_key(	2✔
145	eq_filters,	2✔
146	Some(&SortedInvertedRangeQuery {	2✔
147	field_index: range_query.field_index,	2✔
148	operator_and_value: Some((Operator::LT, Field::Null)),	2✔
149	sort_direction: range_query.sort_direction,	2✔
150	}),	2✔
151	is_single_field_sorted_inverted,	2✔
152	)	2✔
153	.expect("we provided a range query");	2✔
154	match range_query.sort_direction {	2✔
155	SortDirection::Ascending => RangeSpec {	2✔
156	start: Some(KeyEndpoint::Excluding(comparison_key)),	2✔
157	end: Some(KeyEndpoint::Including(null_key)),	2✔
158	direction: SortDirection::Ascending,	2✔
159	},	2✔
160	SortDirection::Descending => RangeSpec {	×
161	start: Some(KeyEndpoint::Including(null_key)),	×
162	end: Some(KeyEndpoint::Excluding(comparison_key)),	×
163	direction: SortDirection::Descending,	×
164	},	×
165	}
166	} else {
167	// Just all of them.
168	RangeSpec {	2✔
169	start: None,	2✔
170	end: None,	2✔
171	direction: range_query.sort_direction,	2✔
172	}	2✔
173	}
174	}
175	}
176	} else {
177	// Here we respond to case 3, examples are `a = 1` or `a = 1 && b = 2`.
178	let comparison_key = comparison_key	89✔
179	.expect("here's at least a eq filter because there's no range query");	89✔
180	RangeSpec {	89✔
181	start: Some(KeyEndpoint::Including(comparison_key.clone())),	89✔
182	end: Some(KeyEndpoint::Including(comparison_key)),	89✔
183	direction: SortDirection::Ascending, // doesn't matter	89✔
184	}	89✔
185	})
186	}
187	IndexScanKind::FullText { filter } => match filter.op {	6✔
188	Operator::Contains => {
189	let token = match &filter.val {	6✔
190	Field::String(token) => token,	4✔
191	Field::Text(token) => token,	2✔
192	_ => return Err(CacheError::Index(IndexError::ExpectedStringFullText)),	×
193	};
194	let key = index::get_full_text_secondary_index(token);	6✔
195	Ok(RangeSpec {	6✔
196	start: Some(KeyEndpoint::Including(key.clone())),	6✔
197	end: Some(KeyEndpoint::Including(key)),	6✔
198	direction: SortDirection::Ascending, // doesn't matter	6✔
199	})	6✔
200	}
201	Operator::MatchesAll \| Operator::MatchesAny => {
202	unimplemented!("matches all and matches any are not implemented")	×
203	}
204	other => panic!("operator {other:?} is not supported by full text index"),	×
205	},
206	}
207	}	119✔
208
209	fn build_sorted_inverted_comparison_key(	135✔
210	eq_filters: &[(usize, Field)],	135✔
211	range_query: Option<&SortedInvertedRangeQuery>,	135✔
212	is_single_field_index: bool,	135✔
213	) -> Option<Vec<u8>> {	135✔
214	let mut fields = vec![];	135✔
215	eq_filters.iter().for_each(\|filter\| {	135✔
216	fields.push(&filter.1);	97✔
217	});	135✔
218	if let Some(range_query) = range_query {	135✔
219	if let Some((_, val)) = &range_query.operator_and_value {	46✔
220	fields.push(val);	42✔
221	}	42✔
222	}	89✔
223	if fields.is_empty() {	135✔
224	None	2✔
225	} else {
226	Some(index::get_secondary_index(&fields, is_single_field_index))	133✔
227	}
228	}	135✔
229
230	/// Here we use the invariant that `null` is greater than anything.
231	fn get_key_interval_from_range_query(	20✔
232	comparison_key: Vec<u8>,	20✔
233	null_key: Vec<u8>,	20✔
234	operator: Operator,	20✔
235	sort_direction: SortDirection,	20✔
236	) -> RangeSpec {	20✔
237	match (operator, sort_direction) {	20✔
238	(Operator::LT, SortDirection::Ascending) => RangeSpec {	4✔
239	start: None,	4✔
240	end: Some(KeyEndpoint::Excluding(comparison_key)),	4✔
241	direction: SortDirection::Ascending,	4✔
242	},	4✔
243	(Operator::LT, SortDirection::Descending) => RangeSpec {	×
244	start: Some(KeyEndpoint::Excluding(comparison_key)),	×
245	end: None,	×
246	direction: SortDirection::Descending,	×
247	},	×
248	(Operator::LTE, SortDirection::Ascending) => RangeSpec {	4✔
249	start: None,	4✔
250	end: Some(KeyEndpoint::Including(comparison_key)),	4✔
251	direction: SortDirection::Ascending,	4✔
252	},	4✔
253	(Operator::LTE, SortDirection::Descending) => RangeSpec {	×
254	start: Some(KeyEndpoint::Including(comparison_key)),	×
255	end: None,	×
256	direction: SortDirection::Descending,	×
257	},	×
258	(Operator::GT, SortDirection::Ascending) => RangeSpec {	8✔
259	start: Some(KeyEndpoint::Excluding(comparison_key)),	8✔
260	end: Some(KeyEndpoint::Excluding(null_key)),	8✔
261	direction: SortDirection::Ascending,	8✔
262	},	8✔
263	(Operator::GT, SortDirection::Descending) => RangeSpec {	2✔
264	start: Some(KeyEndpoint::Excluding(null_key)),	2✔
265	end: Some(KeyEndpoint::Excluding(comparison_key)),	2✔
266	direction: SortDirection::Descending,	2✔
267	},	2✔
268	(Operator::GTE, SortDirection::Ascending) => RangeSpec {	2✔
269	start: Some(KeyEndpoint::Including(comparison_key)),	2✔
270	end: Some(KeyEndpoint::Excluding(null_key)),	2✔
271	direction: SortDirection::Ascending,	2✔
272	},	2✔
273	(Operator::GTE, SortDirection::Descending) => RangeSpec {	×
274	start: Some(KeyEndpoint::Excluding(null_key)),	×
275	end: Some(KeyEndpoint::Including(comparison_key)),	×
276	direction: SortDirection::Descending,	×
277	},	×
278	(other, _) => {	×
279	panic!("operator {other:?} is not supported by sorted inverted index range query")	×
280	}
281	}
282	}	20✔

getdozer / dozer / 4698406303

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