4382731818

Build Type

push

github

Committed by GitHub

Commit Message

fix: Fix publication slot creation (#1202)

Run Details

60 of 60 new or added lines in 24 files covered. (100.0%)

27881 of 39907 relevant lines covered (69.86%)

58951.75 hits per line

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

94.05

/dozer-cache/src/cache/index/mod.rs

use std::cmp::Ordering;

use dozer_types::types::{FieldBorrow, IndexDefinition, Record};

pub trait CacheIndex {
    // Builds one index based on index definition and record
    fn build(index: &IndexDefinition, rec: &Record) -> Vec<Vec<u8>>;

    fn get_key(schema_id: u32, field_idx: &usize, field_val: &[u8]) -> Vec<u8>;
}

use dozer_types::types::Field;

use crate::errors::CompareError;

pub fn get_primary_key(primary_index: &[usize], values: &[Field]) -> Vec<u8> {
    debug_assert!(
        !primary_index.is_empty(),
        "Primary key indexes cannot be empty"
    );

    let key: Vec<Vec<u8>> = primary_index
        .iter()
        .map(|idx| values[*idx].encode())
        .collect();

    key.join("#".as_bytes())
}

/// Returns the secondary index key for a given set of fields.
///
/// We allow computing the secondary index key of "prefix" fields, so the user can filter the "prefix" fields using `Eq` filters,
/// and sort the filtering result using other fields.
///
/// In the meantime, we compute the key differently for single field indexes and compound indexes.
/// We'are not able to tell if certain `fields` belong to a single field index or compound index if its length is 1, hence the second parameter.
///
/// # Parameters
/// - `fields`: The fields to index.
/// - `is_single_field_index`: Whether the `fields` belong to a single field index. If `true`, `fields` must have length 1.
pub fn get_secondary_index(fields: &[&Field], is_single_field_index: bool) -> Vec<u8> {
    debug_assert!(!is_single_field_index || fields.len() == 1);
    if is_single_field_index {
        fields[0].encode()
    } else {
        get_composite_secondary_index(fields)
    }
}

pub fn compare_composite_secondary_index(a: &[u8], b: &[u8]) -> Result<Ordering, CompareError> {
    let mut a = CompositeSecondaryIndexKey::new(a);
    let mut b = CompositeSecondaryIndexKey::new(b);
    Ok(loop {
        match (a.next(), b.next()) {
            (Some(a), Some(b)) => {
                let a = a?;
                let b = b?;
                match a.cmp(&b) {
                    Ordering::Equal => continue,
                    ordering => break ordering,
                }
            }
            (Some(_), None) => break Ordering::Greater,
            (None, Some(_)) => break Ordering::Less,
            (None, None) => break Ordering::Equal,
        }
    })
}

pub fn get_full_text_secondary_index(token: &str) -> Vec<u8> {
    token.as_bytes().to_vec()
}

fn get_composite_secondary_index(fields: &[&Field]) -> Vec<u8> {
    fn get_field_encoding_len(field: &Field) -> usize {
        8 + field.encoding_len()
    }

    let total_len = fields
        .iter()
        .map(|field| get_field_encoding_len(field))
        .sum::<usize>();
    let mut buf = vec![0; total_len];
    let mut offset = 0;
    for field in fields {
        let field_len = get_field_encoding_len(field);
        buf[offset..offset + 8].copy_from_slice(&(field_len as u64).to_be_bytes());
        field.encode_buf(&mut buf[offset + 8..offset + field_len]);
        offset += field_len;
    }
    buf
}

struct CompositeSecondaryIndexKey<'a> {
    buf: &'a [u8],
    offset: usize,
}

impl<'a> CompositeSecondaryIndexKey<'a> {
    fn new(buf: &'a [u8]) -> Self {
        Self { buf, offset: 0 }
    }

    fn decode_one(&mut self) -> Result<FieldBorrow<'a>, CompareError> {
        if self.offset + 8 > self.buf.len() {
            return Err(CompareError::CannotReadFieldLength);
        }

        let field_len = u64::from_be_bytes(
            self.buf[self.offset..self.offset + 8]
                .try_into()
                .expect("We have checked `buf` is long enough"),
        ) as usize;
        if self.offset + field_len > self.buf.len() {
            return Err(CompareError::CannotReadField);
        }

        let field = Field::decode_borrow(&self.buf[self.offset + 8..self.offset + field_len])?;
        self.offset += field_len;
        Ok(field)
    }
}

impl<'a> Iterator for CompositeSecondaryIndexKey<'a> {
    type Item = Result<FieldBorrow<'a>, CompareError>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.offset >= self.buf.len() {
            return None;
        }

        let result = self.decode_one();
        if result.is_err() {
            // Once an error happens, we stop decoding the rest of the buffer.
            self.offset = self.buf.len();
        }
        Some(result)
    }
}

#[cfg(test)]
mod tests;

1	use std::cmp::Ordering;
2
3	use dozer_types::types::{FieldBorrow, IndexDefinition, Record};
4
5	pub trait CacheIndex {
6	// Builds one index based on index definition and record
7	fn build(index: &IndexDefinition, rec: &Record) -> Vec<Vec<u8>>;
8
9	fn get_key(schema_id: u32, field_idx: &usize, field_val: &[u8]) -> Vec<u8>;
10	}
11
12	use dozer_types::types::Field;
13
14	use crate::errors::CompareError;
15
16	pub fn get_primary_key(primary_index: &[usize], values: &[Field]) -> Vec<u8> {	23,949✔
17	debug_assert!(
18	!primary_index.is_empty(),	23,949✔
19	"Primary key indexes cannot be empty"	×
20	);
21
22	let key: Vec<Vec<u8>> = primary_index	23,949✔
23	.iter()	23,949✔
24	.map(\|idx\| values[*idx].encode())	23,949✔
25	.collect();	23,949✔
26		23,949✔
27	key.join("#".as_bytes())	23,949✔
28	}	23,949✔
29
30	/// Returns the secondary index key for a given set of fields.
31	///
32	/// We allow computing the secondary index key of "prefix" fields, so the user can filter the "prefix" fields using `Eq` filters,
33	/// and sort the filtering result using other fields.
34	///
35	/// In the meantime, we compute the key differently for single field indexes and compound indexes.
36	/// We'are not able to tell if certain `fields` belong to a single field index or compound index if its length is 1, hence the second parameter.
37	///
38	/// # Parameters
39	/// - `fields`: The fields to index.
40	/// - `is_single_field_index`: Whether the `fields` belong to a single field index. If `true`, `fields` must have length 1.
41	pub fn get_secondary_index(fields: &[&Field], is_single_field_index: bool) -> Vec<u8> {	168,889✔
42	debug_assert!(!is_single_field_index \|\| fields.len() == 1);	168,973✔
43	if is_single_field_index {	168,937✔
44	fields[0].encode()	112,724✔
45	} else {
46	get_composite_secondary_index(fields)	56,213✔
47	}
48	}	168,937✔
49
50	pub fn compare_composite_secondary_index(a: &[u8], b: &[u8]) -> Result<Ordering, CompareError> {	2,784,501✔
51	let mut a = CompositeSecondaryIndexKey::new(a);	2,784,501✔
52	let mut b = CompositeSecondaryIndexKey::new(b);	2,784,501✔
53	Ok(loop {
54	match (a.next(), b.next()) {	5,906,829✔
55	(Some(a), Some(b)) => {	5,299,011✔
56	let a = a?;	5,299,011✔
57	let b = b?;	5,299,011✔
58	match a.cmp(&b) {	5,299,011✔
59	Ordering::Equal => continue,	3,122,328✔
60	ordering => break ordering,	2,176,683✔
61	}
62	}
63	(Some(_), None) => break Ordering::Greater,	596,792✔
64	(None, Some(_)) => break Ordering::Less,	9,994✔
65	(None, None) => break Ordering::Equal,	1,032✔
66	}
67	})
68	}	2,784,501✔
69
70	pub fn get_full_text_secondary_index(token: &str) -> Vec<u8> {	44,777✔
71	token.as_bytes().to_vec()	44,777✔
72	}	44,777✔
73
74	fn get_composite_secondary_index(fields: &[&Field]) -> Vec<u8> {	57,882✔
75	fn get_field_encoding_len(field: &Field) -> usize {	254,911✔
76	8 + field.encoding_len()	254,911✔
77	}	254,911✔
78		57,882✔
79	let total_len = fields	57,882✔
80	.iter()	57,882✔
81	.map(\|field\| get_field_encoding_len(field))	127,499✔
82	.sum::<usize>();	57,882✔
83	let mut buf = vec![0; total_len];	57,882✔
84	let mut offset = 0;	57,882✔
85	for field in fields {	185,381✔
86	let field_len = get_field_encoding_len(field);	127,499✔
87	buf[offset..offset + 8].copy_from_slice(&(field_len as u64).to_be_bytes());	127,499✔
88	field.encode_buf(&mut buf[offset + 8..offset + field_len]);	127,499✔
89	offset += field_len;	127,499✔
90	}	127,499✔
91	buf	57,882✔
92	}	57,882✔
93
94	struct CompositeSecondaryIndexKey<'a> {
95	buf: &'a [u8],
96	offset: usize,
97	}
98
99	impl<'a> CompositeSecondaryIndexKey<'a> {
100	fn new(buf: &'a [u8]) -> Self {	5,569,508✔
101	Self { buf, offset: 0 }	5,569,508✔
102	}	5,569,508✔
103
104	fn decode_one(&mut self) -> Result<FieldBorrow<'a>, CompareError> {	11,205,796✔
105	if self.offset + 8 > self.buf.len() {	11,205,796✔
106	return Err(CompareError::CannotReadFieldLength);	×
107	}	11,205,796✔
108		11,205,796✔
109	let field_len = u64::from_be_bytes(	11,205,796✔
110	self.buf[self.offset..self.offset + 8]	11,205,796✔
111	.try_into()	11,205,796✔
112	.expect("We have checked `buf` is long enough"),	11,205,796✔
113	) as usize;	11,205,796✔
114	if self.offset + field_len > self.buf.len() {	11,205,796✔
115	return Err(CompareError::CannotReadField);	×
116	}	11,205,796✔
117
118	let field = Field::decode_borrow(&self.buf[self.offset + 8..self.offset + field_len])?;	11,205,798✔
119	self.offset += field_len;	11,205,798✔
120	Ok(field)	11,205,798✔
121	}	11,205,798✔
122	}
123
124	impl<'a> Iterator for CompositeSecondaryIndexKey<'a> {
125	type Item = Result<FieldBorrow<'a>, CompareError>;
126
127	fn next(&mut self) -> Option<Self::Item> {	11,815,154✔
128	if self.offset >= self.buf.len() {	11,815,154✔
129	return None;	609,356✔
130	}	11,205,798✔
131		11,205,798✔
132	let result = self.decode_one();	11,205,798✔
133	if result.is_err() {	11,205,798✔
134	// Once an error happens, we stop decoding the rest of the buffer.	×
135	self.offset = self.buf.len();	×
136	}	11,205,798✔
137	Some(result)	11,205,798✔
138	}	11,815,154✔
139	}
140
141	#[cfg(test)]
142	mod tests;

getdozer / dozer / 4382731818

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