1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use std::borrow::Cow;

use super::{raw, RleDecoder, RleEncoder, Sink};

/// Encodes integers as the change since the previous value.
///
/// The initial value is 0 encoded as u64. Deltas are encoded as i64.
///
/// Run length encoding is then applied to the resulting sequence.
pub(crate) struct DeltaEncoder<S> {
    rle: RleEncoder<S, i64>,
    absolute_value: i64,
}

impl<S: Sink> DeltaEncoder<S> {
    pub(crate) fn new(output: S) -> DeltaEncoder<S> {
        DeltaEncoder {
            rle: RleEncoder::new(output),
            absolute_value: 0,
        }
    }

    pub(crate) fn append_value(&mut self, value: i64) {
        self.rle
            .append_value(value.saturating_sub(self.absolute_value));
        self.absolute_value = value;
    }

    pub(crate) fn append_null(&mut self) {
        self.rle.append_null();
    }

    pub(crate) fn append(&mut self, val: Option<i64>) {
        match val {
            Some(v) => self.append_value(v),
            None => self.append_null(),
        }
    }

    pub(crate) fn finish(self) -> (S, usize) {
        self.rle.finish()
    }
}

impl<S: Sink> From<S> for DeltaEncoder<S> {
    fn from(output: S) -> Self {
        DeltaEncoder::new(output)
    }
}

/// See discussion on [`DeltaEncoder`] for the format data is stored in.
#[derive(Debug, Clone)]
pub(crate) struct DeltaDecoder<'a> {
    rle: RleDecoder<'a, i64>,
    absolute_val: i64,
}

impl<'a> DeltaDecoder<'a> {
    pub(crate) fn done(&self) -> bool {
        self.rle.done()
    }
}

impl<'a> From<Cow<'a, [u8]>> for DeltaDecoder<'a> {
    fn from(bytes: Cow<'a, [u8]>) -> Self {
        DeltaDecoder {
            rle: RleDecoder::from(bytes),
            absolute_val: 0,
        }
    }
}

impl<'a> From<&'a [u8]> for DeltaDecoder<'a> {
    fn from(d: &'a [u8]) -> Self {
        Cow::Borrowed(d).into()
    }
}

impl<'a> Iterator for DeltaDecoder<'a> {
    type Item = Result<Option<i64>, raw::Error>;

    fn next(&mut self) -> Option<Self::Item> {
        match self.rle.next() {
            Some(Ok(next)) => match next {
                Some(delta) => {
                    self.absolute_val = self.absolute_val.saturating_add(delta);
                    Some(Ok(Some(self.absolute_val)))
                }
                None => Some(Ok(None)),
            },
            Some(Err(e)) => Some(Err(e)),
            None => None,
        }
    }
}