Support for plain encoded binary in data pages

src/encoding/mod.rs

@@ -5,6 +5,7 @@ pub mod delta_bitpacked;
 pub mod delta_byte_array;
 pub mod delta_length_byte_array;
 pub mod hybrid_rle;
+pub mod plain_byte_array;
 pub mod uleb128;
 pub mod zigzag_leb128;
 

src/encoding/plain_byte_array/decoder.rs

@@ -0,0 +1,46 @@
+use crate::encoding::get_length;
+
+/// Decodes according to [Plain strings](https://github.com/apache/parquet-format/blob/master/Encodings.md#plain-plain--0),
+/// prefixes, lengths and values
+/// # Implementation
+/// This struct does not allocate on the heap.
+#[derive(Debug)]
+pub struct Decoder<'a> {
+    values: &'a [u8],
+    remaining: usize,
+}
+
+impl<'a> Decoder<'a> {
+    #[inline]
+    pub fn new(values: &'a [u8], length: usize) -> Self {
+        Self {
+            values,
+            remaining: length,
+        }
+    }
+}
+
+impl<'a> Iterator for Decoder<'a> {
+    type Item = &'a [u8];
+
+    #[inline]
+    fn next(&mut self) -> Option<Self::Item> {
+        let values = self.values;
+        if values.len() >= 4 {
+            let next_len = get_length(values) as usize;
+            let values = &values[4..];
+
+            let result = Some(&values[0..next_len]);
+            self.values = &values[next_len..];
+            self.remaining -= 1;
+
+            result
+        } else {
+            None
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (self.remaining, None)
+    }
+}

src/encoding/plain_byte_array/mod.rs

@@ -0,0 +1,3 @@
+mod decoder;
+
+pub use decoder::Decoder;

src/serialization/read/binary.rs

@@ -4,8 +4,9 @@ use super::levels::consume_level;
 use crate::error::{ParquetError, Result};
 use crate::metadata::ColumnDescriptor;
 use crate::read::Page;
+use crate::serialization::read::utils::ValuesDef;
 use crate::{
-    encoding::{bitpacking, uleb128},
+    encoding::{bitpacking, plain_byte_array, uleb128},
     read::BinaryPageDict,
 };
 
@@ -62,3 +63,39 @@ pub fn page_dict_to_vec(
         _ => todo!(),
     }
 }
+
+fn read_plain_buffer(
+    values: &[u8],
+    length: u32,
+    def_level_encoding: (&Encoding, i16),
+) -> Vec<Option<Vec<u8>>> {
+    let (values, def_levels) = consume_level(values, length, def_level_encoding);
+
+    let decoded_values =
+        plain_byte_array::Decoder::new(values, length as usize).map(|bytes| bytes.to_vec());
+
+    ValuesDef::new(
+        decoded_values,
+        def_levels.into_iter(),
+        def_level_encoding.1 as u32,
+    )
+    .collect()
+}
+
+pub fn page_to_vec(page: &Page, descriptor: &ColumnDescriptor) -> Result<Vec<Option<Vec<u8>>>> {
+    assert_eq!(descriptor.max_rep_level(), 0);
+    match page {
+        Page::V1(page) => match (&page.header.encoding, &page.dictionary_page) {
+            (Encoding::Plain, None) => Ok(read_plain_buffer(
+                &page.buffer,
+                page.header.num_values as u32,
+                (
+                    &page.header.definition_level_encoding,
+                    descriptor.max_def_level(),
+                ),
+            )),
+            _ => todo!(),
+        },
+        _ => todo!(),
+    }
+}

src/serialization/read/mod.rs

@@ -88,6 +88,9 @@ pub fn page_to_array(page: CompressedPage, descriptor: &ColumnDescriptor) -> Res
                 PhysicalType::Double => {
                     Ok(Array::Float64(primitive::page_to_vec(&page, descriptor)?))
                 }
+                PhysicalType::ByteArray => {
+                    Ok(Array::Binary(binary::page_to_vec(&page, descriptor)?))
+                }
                 _ => todo!(),
             },
         },