6105410942

Committed 07 Sep 2023 04:28AM UTC coverage: 77.562% (-0.1%) from 77.686%

Build # 6105410942

Build Type

push

github

Committed by

chloeminkyung

Commit Message

feat: onnx image

Run Details

1141 of 1141 new or added lines in 66 files covered. (100.0%)

49957 of 64409 relevant lines covered (77.56%)

50900.25 hits per line

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

0.0

/dozer-sql/src/pipeline/utils/serialize.rs

use dozer_core::{
    dozer_log::{storage::Object, tokio::sync::mpsc::error::SendError},
    processor_record::{ProcessorRecord, ProcessorRecordStore},
};
use dozer_types::{
    bincode,
    serde::{de::DeserializeOwned, Serialize},
    thiserror::{self, Error},
};

#[derive(Debug, Error)]
pub enum SerializationError {
    #[error("bincode error: {0}")]
    Bincode(#[from] bincode::Error),
    #[error("Cannot send value to persisting thread")]
    SendError,
}

impl<T> From<SendError<T>> for SerializationError {
    fn from(_: SendError<T>) -> Self {
        Self::SendError
    }
}

pub struct Cursor<'a>(&'a [u8]);

impl<'a> Cursor<'a> {
    pub fn new(data: &'a [u8]) -> Self {
        Self(data)
    }

    pub fn consume(&mut self, len: usize) -> Result<&'a [u8], DeserializationError> {
        if self.0.len() < len {
            return Err(DeserializationError::NotEnoughData {
                requested: len,
                remaining: self.0.len(),
            });
        }

        let (head, tail) = self.0.split_at(len);
        self.0 = tail;
        Ok(head)
    }
}

#[derive(Debug, Error)]
pub enum DeserializationError {
    #[error("not enough data: requested {requested}, remaining {remaining}")]
    NotEnoughData { requested: usize, remaining: usize },
    #[error("bincode error: {0}")]
    Bincode(#[from] bincode::Error),
}

pub fn serialize_u64(value: u64, object: &mut Object) -> Result<(), SerializationError> {
    object.write(&value.to_le_bytes()).map_err(Into::into)
}

pub fn deserialize_u64(cursor: &mut Cursor) -> Result<u64, DeserializationError> {
    let bytes = cursor.consume(8)?;
    Ok(u64::from_le_bytes(bytes.try_into().unwrap()))
}

pub fn serialize_vec_u8(vec: &[u8], object: &mut Object) -> Result<(), SerializationError> {
    serialize_u64(vec.len() as u64, object)?;
    object.write(vec).map_err(Into::into)
}

pub fn deserialize_vec_u8<'a>(cursor: &mut Cursor<'a>) -> Result<&'a [u8], DeserializationError> {
    let len = deserialize_u64(cursor)? as usize;
    let data = cursor.consume(len)?;
    Ok(data)
}

pub fn serialize_bincode(
    value: impl Serialize,
    object: &mut Object,
) -> Result<(), SerializationError> {
    let data = bincode::serialize(&value)?;
    serialize_vec_u8(&data, object)
}

pub fn deserialize_bincode<T: DeserializeOwned>(
    cursor: &mut Cursor,
) -> Result<T, DeserializationError> {
    let data = deserialize_vec_u8(cursor)?;
    Ok(bincode::deserialize(data)?)
}

pub fn serialize_record(
    record: &ProcessorRecord,
    record_store: &ProcessorRecordStore,
    object: &mut Object,
) -> Result<(), SerializationError> {
    serialize_vec_u8(&record_store.serialize_record(record)?, object)
}

pub fn deserialize_record(
    cursor: &mut Cursor,
    record_store: &ProcessorRecordStore,
) -> Result<ProcessorRecord, DeserializationError> {
    record_store
        .deserialize_record(deserialize_vec_u8(cursor)?)
        .map_err(Into::into)
}

1	use dozer_core::{
2	dozer_log::{storage::Object, tokio::sync::mpsc::error::SendError},
3	processor_record::{ProcessorRecord, ProcessorRecordStore},
4	};
5	use dozer_types::{
6	bincode,
7	serde::{de::DeserializeOwned, Serialize},
8	thiserror::{self, Error},
9	};
10
11	#[derive(Debug, Error)]	×
12	pub enum SerializationError {	×
13	#[error("bincode error: {0}")]
14	Bincode(#[from] bincode::Error),
15	#[error("Cannot send value to persisting thread")]
16	SendError,
17	}
18
19	impl<T> From<SendError<T>> for SerializationError {
20	fn from(_: SendError<T>) -> Self {	×
21	Self::SendError	×
22	}	×
23	}
24
25	pub struct Cursor<'a>(&'a [u8]);
26
27	impl<'a> Cursor<'a> {
28	pub fn new(data: &'a [u8]) -> Self {	×
29	Self(data)	×
30	}	×
31
32	pub fn consume(&mut self, len: usize) -> Result<&'a [u8], DeserializationError> {	×
33	if self.0.len() < len {	×
34	return Err(DeserializationError::NotEnoughData {	×
35	requested: len,	×
36	remaining: self.0.len(),	×
37	});	×
38	}	×
39		×
40	let (head, tail) = self.0.split_at(len);	×
41	self.0 = tail;	×
42	Ok(head)	×
43	}	×
44	}
45
46	#[derive(Debug, Error)]	×
47	pub enum DeserializationError {	×
48	#[error("not enough data: requested {requested}, remaining {remaining}")]
49	NotEnoughData { requested: usize, remaining: usize },
50	#[error("bincode error: {0}")]
51	Bincode(#[from] bincode::Error),
52	}
53
54	pub fn serialize_u64(value: u64, object: &mut Object) -> Result<(), SerializationError> {	×
55	object.write(&value.to_le_bytes()).map_err(Into::into)	×
56	}	×
57
58	pub fn deserialize_u64(cursor: &mut Cursor) -> Result<u64, DeserializationError> {	×
59	let bytes = cursor.consume(8)?;	×
60	Ok(u64::from_le_bytes(bytes.try_into().unwrap()))	×
61	}	×
62
63	pub fn serialize_vec_u8(vec: &[u8], object: &mut Object) -> Result<(), SerializationError> {
64	serialize_u64(vec.len() as u64, object)?;	×
65	object.write(vec).map_err(Into::into)	×
66	}	×
67
68	pub fn deserialize_vec_u8<'a>(cursor: &mut Cursor<'a>) -> Result<&'a [u8], DeserializationError> {	×
69	let len = deserialize_u64(cursor)? as usize;	×
70	let data = cursor.consume(len)?;	×
71	Ok(data)	×
72	}	×
73
74	pub fn serialize_bincode(	×
75	value: impl Serialize,	×
76	object: &mut Object,	×
77	) -> Result<(), SerializationError> {	×
78	let data = bincode::serialize(&value)?;	×
79	serialize_vec_u8(&data, object)	×
80	}	×
81
82	pub fn deserialize_bincode<T: DeserializeOwned>(	×
83	cursor: &mut Cursor,	×
84	) -> Result<T, DeserializationError> {	×
85	let data = deserialize_vec_u8(cursor)?;	×
86	Ok(bincode::deserialize(data)?)	×
87	}	×
88
89	pub fn serialize_record(	×
90	record: &ProcessorRecord,	×
91	record_store: &ProcessorRecordStore,	×
92	object: &mut Object,	×
93	) -> Result<(), SerializationError> {	×
94	serialize_vec_u8(&record_store.serialize_record(record)?, object)	×
95	}	×
96
97	pub fn deserialize_record(
98	cursor: &mut Cursor,
99	record_store: &ProcessorRecordStore,
100	) -> Result<ProcessorRecord, DeserializationError> {
101	record_store
102	.deserialize_record(deserialize_vec_u8(cursor)?)	×
103	.map_err(Into::into)	×
104	}	×

getdozer / dozer / 6105410942

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