reifydb_value/value/digest/
encoding.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::collections::BTreeMap;
5
6use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
7use serde_bytes::ByteBuf;
8
9use crate::value::{
10	digest::{Digest, DigestError, inner_from_tag, inner_tag},
11	value_type::ValueType,
12};
13
14const VERSION: u64 = 1;
15
16impl Digest {
17	pub fn encode(&self) -> Vec<u8> {
18		let mut out = Vec::new();
19		write_varint(&mut out, VERSION);
20		write_varint(&mut out, u64::from(inner_tag(&self.inner).expect("digest inner type has a tag")));
21		write_varint(&mut out, u64::from(self.accuracy));
22		write_varint(&mut out, self.zero);
23		write_varint(&mut out, self.negative_infinity);
24		write_varint(&mut out, self.positive_infinity);
25		write_store(&mut out, &self.negative);
26		write_store(&mut out, &self.positive);
27		out
28	}
29
30	pub fn decode(bytes: &[u8]) -> Result<Self, DigestError> {
31		let mut reader = Reader {
32			bytes,
33			position: 0,
34		};
35		let version = reader.varint()?;
36		if version != VERSION {
37			return Err(DigestError::UnknownVersion {
38				version,
39			});
40		}
41		let tag = reader.varint()?;
42		let inner = inner_from_tag(tag).ok_or(DigestError::UnknownInnerTag {
43			tag,
44		})?;
45		let accuracy = u32::try_from(reader.varint()?).map_err(|_| DigestError::AccuracyOutOfRange)?;
46		let mut digest = Digest::new(inner, accuracy)?;
47		digest.zero = reader.varint()?;
48		digest.negative_infinity = reader.varint()?;
49		digest.positive_infinity = reader.varint()?;
50		digest.negative = reader.store()?;
51		digest.positive = reader.store()?;
52		if reader.position != bytes.len() {
53			return Err(DigestError::TrailingBytes {
54				remaining: bytes.len() - reader.position,
55			});
56		}
57		if digest.inner == ValueType::Duration
58			&& (digest.negative_infinity != 0 || digest.positive_infinity != 0)
59		{
60			return Err(DigestError::NonFiniteDuration);
61		}
62		if digest.checked_count().is_none() {
63			return Err(DigestError::CountOverflow);
64		}
65		Ok(digest)
66	}
67}
68
69impl Serialize for Digest {
70	fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
71	where
72		S: Serializer,
73	{
74		serializer.serialize_bytes(&self.encode())
75	}
76}
77
78impl<'de> Deserialize<'de> for Digest {
79	fn deserialize<D>(deserializer: D) -> Result<Digest, D::Error>
80	where
81		D: Deserializer<'de>,
82	{
83		let bytes = ByteBuf::deserialize(deserializer)?;
84		Digest::decode(&bytes).map_err(de::Error::custom)
85	}
86}
87
88fn write_varint(out: &mut Vec<u8>, mut value: u64) {
89	while value >= 0x80 {
90		out.push((value as u8) | 0x80);
91		value >>= 7;
92	}
93	out.push(value as u8);
94}
95
96fn write_store(out: &mut Vec<u8>, store: &BTreeMap<i32, u64>) {
97	write_varint(out, store.len() as u64);
98	let mut previous: Option<i32> = None;
99	for (&index, &count) in store {
100		match previous {
101			None => write_varint(out, zigzag(index)),
102			Some(previous) => write_varint(out, (i64::from(index) - i64::from(previous)) as u64),
103		}
104		write_varint(out, count);
105		previous = Some(index);
106	}
107}
108
109fn zigzag(index: i32) -> u64 {
110	u64::from(((index << 1) ^ (index >> 31)) as u32)
111}
112
113fn unzigzag(encoded: u32) -> i32 {
114	((encoded >> 1) as i32) ^ -((encoded & 1) as i32)
115}
116
117struct Reader<'a> {
118	bytes: &'a [u8],
119	position: usize,
120}
121
122impl Reader<'_> {
123	fn varint(&mut self) -> Result<u64, DigestError> {
124		let mut value = 0u64;
125		let mut shift = 0u32;
126		loop {
127			let byte = *self.bytes.get(self.position).ok_or(DigestError::Truncated)?;
128			self.position += 1;
129			let low = u64::from(byte & 0x7f);
130			if shift == 63 && low > 1 {
131				return Err(DigestError::VarintOverflow);
132			}
133			value |= low << shift;
134			if byte & 0x80 == 0 {
135				if byte == 0 && shift > 0 {
136					return Err(DigestError::NonMinimalVarint);
137				}
138				return Ok(value);
139			}
140			shift += 7;
141			if shift > 63 {
142				return Err(DigestError::VarintOverflow);
143			}
144		}
145	}
146
147	fn store(&mut self) -> Result<BTreeMap<i32, u64>, DigestError> {
148		let len = self.varint()?;
149		let mut store = BTreeMap::new();
150		let mut previous: Option<i32> = None;
151		for _ in 0..len {
152			let index = match previous {
153				None => {
154					let encoded = u32::try_from(self.varint()?)
155						.map_err(|_| DigestError::BucketIndexOutOfRange)?;
156					unzigzag(encoded)
157				}
158				Some(previous) => {
159					let delta = self.varint()?;
160					if delta == 0 {
161						return Err(DigestError::RepeatedBucketIndex {
162							index: previous,
163						});
164					}
165					i64::try_from(delta)
166						.ok()
167						.and_then(|delta| i64::from(previous).checked_add(delta))
168						.and_then(|index| i32::try_from(index).ok())
169						.ok_or(DigestError::BucketIndexOutOfRange)?
170				}
171			};
172			let count = self.varint()?;
173			if count == 0 {
174				return Err(DigestError::ZeroBucketCount {
175					index,
176				});
177			}
178			store.insert(index, count);
179			previous = Some(index);
180		}
181		Ok(store)
182	}
183}
184
185#[cfg(test)]
186mod tests {
187	use postcard::{from_bytes, to_allocvec};
188	use serde_bytes::Bytes;
189	use serde_json::{from_str, to_string};
190
191	use super::*;
192	use crate::value::digest::{
193		INNER_TAGS,
194		tests::{Rng, any_value, built, float_digest},
195	};
196
197	const EMPTY_FLOAT8: [u64; 8] = [1, 3, 10_000, 0, 0, 0, 0, 0];
198
199	fn varints(values: &[u64]) -> Vec<u8> {
200		let mut out = Vec::new();
201		for &value in values {
202			write_varint(&mut out, value);
203		}
204		out
205	}
206
207	fn header_with(position: usize, value: u64) -> Vec<u64> {
208		let mut fields = EMPTY_FLOAT8[..6].to_vec();
209		fields[position] = value;
210		fields
211	}
212
213	fn with_positive(positive: &[u64]) -> Vec<u8> {
214		varints(&[&EMPTY_FLOAT8[..7], positive].concat())
215	}
216
217	fn with_negative(negative: &[u64]) -> Vec<u8> {
218		varints(&[&EMPTY_FLOAT8[..6], negative, &[0]].concat())
219	}
220
221	fn sample() -> Digest {
222		built(10_000, &[-2.0, 0.0, 0.5, 1.0, 1.0, 2.0, f64::INFINITY])
223	}
224
225	fn shuffle(rng: &mut Rng, values: &mut [f64]) {
226		for i in (1..values.len()).rev() {
227			values.swap(i, rng.below(i as u64 + 1) as usize);
228		}
229	}
230
231	#[test]
232	fn canonical_bytes_are_pinned() {
233		// Stored digests are read back with these exact bytes; any layout change needs a new version.
234		assert_eq!(
235			sample().encode(),
236			vec![
237				0x01, 0x03, 0x90, 0x4e, 0x01, 0x00, 0x01, 0x01, 0x46, 0x01, 0x03, 0x43, 0x01, 0x22,
238				0x02, 0x23, 0x01
239			]
240		);
241		assert_eq!(Digest::decode(&sample().encode()).unwrap(), sample());
242		assert_eq!(Digest::decode(&varints(&EMPTY_FLOAT8)).unwrap(), float_digest(10_000));
243	}
244
245	#[test]
246	fn inner_type_tags_are_the_frozen_type_numbers() {
247		let pinned = [
248			(ValueType::Float4, 2),
249			(ValueType::Float8, 3),
250			(ValueType::Int1, 4),
251			(ValueType::Int2, 5),
252			(ValueType::Int4, 6),
253			(ValueType::Int8, 7),
254			(ValueType::Int16, 8),
255			(ValueType::Uint1, 10),
256			(ValueType::Uint2, 11),
257			(ValueType::Uint4, 12),
258			(ValueType::Uint8, 13),
259			(ValueType::Uint16, 14),
260			(ValueType::Duration, 18),
261		];
262		for (inner, tag) in pinned {
263			let digest = Digest::new(inner.clone(), 10_000).unwrap();
264			assert_eq!(digest.encode()[1], tag, "{inner}");
265			assert_eq!(Digest::decode(&digest.encode()).unwrap().inner(), &inner);
266		}
267	}
268
269	#[test]
270	fn varint_and_zigzag_are_pinned() {
271		let cases: [(u64, &[u8]); 5] = [
272			(0, &[0x00]),
273			(127, &[0x7f]),
274			(128, &[0x80, 0x01]),
275			(300, &[0xac, 0x02]),
276			(u64::MAX, &[0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x01]),
277		];
278		for (value, bytes) in cases {
279			assert_eq!(varints(&[value]), bytes);
280			let mut reader = Reader {
281				bytes,
282				position: 0,
283			};
284			assert_eq!(reader.varint().unwrap(), value);
285		}
286		for (index, encoded) in
287			[(0, 0), (-1, 1), (1, 2), (-34, 67), (i32::MAX, u32::MAX - 1), (i32::MIN, u32::MAX)]
288		{
289			assert_eq!(zigzag(index), u64::from(encoded));
290			assert_eq!(unzigzag(encoded), index);
291		}
292	}
293
294	#[test]
295	fn round_trip_keeps_inner_type_accuracy_and_every_count() {
296		let mut rng = Rng::new(5);
297		for (_, inner) in INNER_TAGS {
298			for accuracy in [1_000, 12_345, 100_000] {
299				let mut digest = Digest::new(inner.clone(), accuracy).unwrap();
300				for _ in 0..500 {
301					let value = any_value(&mut rng);
302					if inner != ValueType::Duration || value.is_finite() {
303						digest.add(value);
304					}
305				}
306				let bytes = digest.encode();
307				assert_eq!(bytes[0], 1, "the version tag must be the first byte");
308				let decoded = Digest::decode(&bytes).unwrap();
309				assert_eq!(decoded, digest);
310				assert_eq!(decoded.inner(), &inner);
311				assert_eq!(decoded.accuracy(), accuracy);
312				assert_eq!(decoded.encode(), bytes);
313			}
314		}
315	}
316
317	#[test]
318	fn any_add_order_gives_equal_bytes() {
319		let mut rng = Rng::new(9);
320		for accuracy in [1_000, 10_000, 100_000] {
321			let mut values: Vec<f64> = (0..1_000).map(|_| any_value(&mut rng)).collect();
322			let bytes = built(accuracy, &values).encode();
323			for _ in 0..5 {
324				shuffle(&mut rng, &mut values);
325				assert_eq!(built(accuracy, &values).encode(), bytes);
326			}
327			let mut churned = built(accuracy, &values);
328			let extra: Vec<f64> = (0..300).map(|_| any_value(&mut rng)).collect();
329			for &value in &extra {
330				churned.add(value);
331			}
332			for &value in extra.iter().rev() {
333				churned.remove(value);
334			}
335			assert_eq!(churned.encode(), bytes, "history must not reach the bytes");
336		}
337	}
338
339	#[test]
340	fn decode_rejects_unknown_version() {
341		for version in [0, 2, 1 << 40] {
342			let bytes = varints(&[&header_with(0, version)[..], &[0, 0]].concat());
343			assert!(
344				matches!(Digest::decode(&bytes), Err(DigestError::UnknownVersion { version: v }) if v == version)
345			);
346		}
347	}
348
349	#[test]
350	fn decode_rejects_trailing_bytes() {
351		let mut bytes = sample().encode();
352		bytes.push(0x00);
353		assert!(matches!(
354			Digest::decode(&bytes),
355			Err(DigestError::TrailingBytes {
356				remaining: 1
357			})
358		));
359	}
360
361	#[test]
362	fn decode_rejects_every_truncation() {
363		let bytes = sample().encode();
364		for len in 0..bytes.len() {
365			assert!(
366				matches!(Digest::decode(&bytes[..len]), Err(DigestError::Truncated)),
367				"prefix of {len} bytes"
368			);
369		}
370		assert!(matches!(Digest::decode(&[0x01, 0x83]), Err(DigestError::Truncated)));
371	}
372
373	#[test]
374	fn decode_rejects_repeated_bucket_index() {
375		assert!(matches!(
376			Digest::decode(&with_positive(&[2, 0, 1, 0, 1])),
377			Err(DigestError::RepeatedBucketIndex {
378				index: 0
379			})
380		));
381		assert!(matches!(
382			Digest::decode(&with_negative(&[3, 9, 1, 1, 1, 0, 1])),
383			Err(DigestError::RepeatedBucketIndex {
384				index: -4
385			})
386		));
387	}
388
389	#[test]
390	fn decode_rejects_indexes_that_leave_i32() {
391		// Deltas are unsigned, so an unsorted store can only be written as an index that runs past i32.
392		assert!(matches!(
393			Digest::decode(&with_positive(&[2, zigzag(i32::MAX), 1, 1, 1])),
394			Err(DigestError::BucketIndexOutOfRange)
395		));
396		assert!(matches!(
397			Digest::decode(&with_positive(&[2, zigzag(i32::MIN), 1, u64::MAX, 1])),
398			Err(DigestError::BucketIndexOutOfRange)
399		));
400		assert!(matches!(
401			Digest::decode(&with_positive(&[1, 1 << 32, 1])),
402			Err(DigestError::BucketIndexOutOfRange)
403		));
404		let widest = Digest::decode(&with_positive(&[2, zigzag(i32::MIN), 1, u64::from(u32::MAX), 1])).unwrap();
405		assert_eq!(widest.bucket_count(), 2);
406	}
407
408	#[test]
409	fn decode_rejects_zero_bucket_count() {
410		assert!(matches!(
411			Digest::decode(&with_positive(&[1, 0, 0])),
412			Err(DigestError::ZeroBucketCount {
413				index: 0
414			})
415		));
416		assert!(matches!(
417			Digest::decode(&with_negative(&[2, 9, 1, 3, 0])),
418			Err(DigestError::ZeroBucketCount {
419				index: -2
420			})
421		));
422	}
423
424	#[test]
425	fn decode_rejects_non_minimal_varints() {
426		let padded_version = [&[0x81, 0x00][..], &varints(&[3, 10_000, 0, 0, 0, 0, 0])].concat();
427		let padded_accuracy = [&varints(&[1, 3])[..], &[0x90, 0xce, 0x00], &varints(&[0, 0, 0, 0, 0])].concat();
428		let padded_count = [&varints(&EMPTY_FLOAT8[..7])[..], &[0x01, 0x00, 0x81, 0x00]].concat();
429		let padded_delta = [&varints(&EMPTY_FLOAT8[..7])[..], &[0x02, 0x00, 0x01, 0x81, 0x00, 0x01]].concat();
430		for bytes in [padded_version, padded_accuracy, padded_count, padded_delta] {
431			assert!(matches!(Digest::decode(&bytes), Err(DigestError::NonMinimalVarint)), "{bytes:02x?}");
432		}
433	}
434
435	#[test]
436	fn decode_rejects_varints_above_u64() {
437		let fields = varints(&EMPTY_FLOAT8[..3]);
438		let ten_continued = [&fields[..], &[0xff; 10], &[0x01]].concat();
439		let tenth_too_high = [&fields[..], &[0xff; 9], &[0x02]].concat();
440		for bytes in [ten_continued, tenth_too_high] {
441			assert!(matches!(Digest::decode(&bytes), Err(DigestError::VarintOverflow)), "{bytes:02x?}");
442		}
443		let largest = varints(&[&header_with(3, u64::MAX)[..], &[0, 0]].concat());
444		assert_eq!(Digest::decode(&largest).unwrap().count(), u64::MAX);
445	}
446
447	#[test]
448	fn decode_rejects_unknown_inner_type_tag() {
449		for tag in [0, 1, 9, 15, 25, 32, 63, 255] {
450			let bytes = varints(&[&header_with(1, tag)[..], &[0, 0]].concat());
451			assert!(
452				matches!(Digest::decode(&bytes), Err(DigestError::UnknownInnerTag { tag: t }) if t == tag)
453			);
454		}
455	}
456
457	#[test]
458	fn decode_rejects_accuracy_outside_range() {
459		for accuracy in [0, 999, 100_001, 1 << 32] {
460			let bytes = varints(&[&header_with(2, accuracy)[..], &[0, 0]].concat());
461			assert!(matches!(Digest::decode(&bytes), Err(DigestError::AccuracyOutOfRange)), "{accuracy}");
462		}
463	}
464
465	#[test]
466	fn decode_rejects_counts_that_overflow_u64() {
467		let specials = varints(&[1, 3, 10_000, u64::MAX, 0, 1, 0, 0]);
468		let buckets = with_positive(&[2, 0, u64::MAX, 1, 1]);
469		for bytes in [specials, buckets] {
470			assert!(matches!(Digest::decode(&bytes), Err(DigestError::CountOverflow)), "{bytes:02x?}");
471		}
472	}
473
474	#[test]
475	fn decode_rejects_infinity_in_a_duration_digest() {
476		for position in [4, 5] {
477			let mut fields = header_with(position, 1);
478			fields[1] = 18;
479			let bytes = varints(&[&fields[..], &[0, 0]].concat());
480			assert!(matches!(Digest::decode(&bytes), Err(DigestError::NonFiniteDuration)), "{position}");
481		}
482		let zero = varints(&[1, 18, 10_000, 1, 0, 0, 0, 0]);
483		assert_eq!(Digest::decode(&zero).unwrap().count(), 1);
484	}
485
486	#[test]
487	fn postcard_carries_exactly_the_encoded_bytes() {
488		let digest = sample();
489		let encoded = digest.encode();
490		let postcard_bytes = to_allocvec(&digest).unwrap();
491		assert_eq!(postcard_bytes, [&varints(&[encoded.len() as u64])[..], &encoded].concat());
492		assert_eq!(from_bytes::<Digest>(&postcard_bytes).unwrap(), digest);
493
494		let framed = to_allocvec(&(7u8, digest.clone(), 9u8)).unwrap();
495		assert_eq!(from_bytes::<(u8, Digest, u8)>(&framed).unwrap(), (7, digest, 9));
496	}
497
498	#[test]
499	fn postcard_rejects_malformed_digest_bytes() {
500		let mut bytes = sample().encode();
501		bytes[0] = 2;
502		let framed = to_allocvec(&Bytes::new(&bytes)).unwrap();
503		assert!(from_bytes::<Digest>(&framed).is_err());
504	}
505
506	#[test]
507	fn json_round_trip() {
508		let digest = sample();
509		let json = to_string(&digest).unwrap();
510		assert_eq!(from_str::<Digest>(&json).unwrap(), digest);
511	}
512}
reifydb_value/value/digest/encoding.rs

reifydb_value/value/digest/
encoding.rs
