Add BYTE_STREAM_SPLIT encoder/decoder (fixes #208)

src/encoding/byte_stream_split/decoder.rs

@@ -0,0 +1,61 @@
+use std::marker::PhantomData;
+use crate::error::Error;
+use crate::types::NativeType;
+
+/// Decodes according to [Byte Stream Split](https://github.com/apache/parquet-format/blob/master/Encodings.md#byte-stream-split-byte_stream_split--9).
+/// # Implementation
+/// This struct does not allocate on the heap.
+#[derive(Debug)]
+pub struct Decoder<'a, T: NativeType> {
+    values: &'a [u8],
+    buffer: Vec<u8>,
+    num_elements: usize,
+    current: usize,
+    element_size: usize,
+    element_type: PhantomData<T>
+}
+
+impl<'a, T: NativeType> Decoder<'a, T> {
+    pub fn try_new(values: &'a [u8]) -> Result<Self, Error> {
+        let element_size = std::mem::size_of::<T>();
+        let values_size = values.len();
+        if values_size % element_size != 0 {
+            return Err(Error::oos("Value array is not a multiple of element size"));
+        }
+        let num_elements = values.len() / element_size;
+        Ok(Self {
+            values,
+            buffer: vec![0_u8; element_size],
+            num_elements,
+            current: 0,
+            element_size,
+            element_type: PhantomData
+        })
+    }
+}
+
+impl<'a, T: NativeType> Iterator for Decoder<'a, T> {
+    type Item = Result<T, Error>;
+
+    #[inline]
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.current >= self.num_elements {
+            return None
+        }
+
+        for n in 0..self.element_size {
+            self.buffer[n] = self.values[(self.num_elements * n) + self.current]
+        }
+
+        let value = T::from_le_bytes(self.buffer.as_slice().try_into().unwrap());
+
+        self.current += 1;
+
+        return Some(Ok(value));
+    }
+
+    #[inline]
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (self.num_elements, Some(self.num_elements))
+    }
+}

src/encoding/byte_stream_split/encoder.rs

@@ -0,0 +1,17 @@
+use crate::types::NativeType;
+
+/// Encodes an array of NativeType according to BYTE_STREAM_SPLIT
+pub fn encode<T: NativeType>(data: &[T], buffer: &mut Vec<u8>) {
+    let element_size = std::mem::size_of::<T>();
+    let num_elements = data.len();
+    let total_length = element_size * num_elements;
+    buffer.resize(total_length, 0);
+
+    for (i, v) in data.iter().enumerate() {
+        let value_bytes = v.to_le_bytes();
+        let value_bytes_ref = value_bytes.as_ref();
+        for n in 0..element_size {
+            buffer[(num_elements * n) + i] = value_bytes_ref[n];
+        }
+    }
+}

src/encoding/byte_stream_split/mod.rs

@@ -0,0 +1,49 @@
+mod decoder;
+mod encoder;
+
+pub use decoder::Decoder;
+pub use encoder::encode;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::error::Error;
+
+    #[test]
+    fn basic() -> Result<(), Error> {
+        let data = vec![1.0_f32, 2.0_f32, 3.0_f32];
+        let mut buffer = vec![];
+        encode(&data, &mut buffer);
+
+        let mut decoder = Decoder::<f32>::try_new(&buffer)?;
+        let values = decoder.by_ref().collect::<Result<Vec<_>, _>>()?;
+
+        assert_eq!(data, values);
+
+        Ok(())
+    }
+
+    #[test]
+    fn pyarrow_integration() -> Result<(), Error> {
+        let buffer = vec![
+            0, 205, 0, 205, 0, 0, 204, 0, 204, 0, 128, 140, 0, 140, 128, 255, 191, 0, 63, 127
+        ];
+
+        let mut decoder = Decoder::<f32>::try_new(&buffer)?;
+        let values = decoder.by_ref().collect::<Result<Vec<_>, _>>()?;
+
+        assert_eq!(values, vec![-f32::INFINITY, -1.1, 0.0, 1.1, f32::INFINITY]);
+
+        Ok(())
+    }
+
+    #[test]
+    fn fails_for_bad_size() -> Result<(), Error> {
+        let buffer = vec![0; 12];
+
+        let result = Decoder::<f64>::try_new(&buffer);
+        assert!(result.is_err());
+
+        Ok(())
+    }
+}

src/encoding/mod.rs

@@ -1,6 +1,7 @@
 use std::convert::TryInto;
 
 pub mod bitpacked;
+pub mod byte_stream_split;
 pub mod delta_bitpacked;
 pub mod delta_byte_array;
 pub mod delta_length_byte_array;