Repository: jorgecarleitao/arrow2 (rust)

Issue:
Reading compressed file produces an error: failed to fill whole buffer
I'm trying new feature added in #566 
and I'm getting an error from Reader: `failed to fill whole buffer`

Here is my sample code:
```
use arrow2::array::Int32Array;
use arrow2::datatypes::{DataType, Field, Schema};
use arrow2::error::Result;
use arrow2::io::ipc::read::{read_file_metadata, FileReader};
use arrow2::io::ipc::write;
use arrow2::io::ipc::write::Compression;
use arrow2::record_batch::RecordBatch;
use std::fs::File;
use std::sync::Arc;

fn main() -> Result<()> {
    let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);

    let a = Int32Array::from_slice(&[0; 10]);
    let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?;

    let file = File::create("test.arrow")?;
    let options = write::WriteOptions {
        compression: Some(Compression::LZ4),
    };
    let mut writer = write::FileWriter::try_new(file, &schema, options)?;
    writer.write(&batch)?;
    writer.finish()?;

    let mut file = File::open("test.arrow")?;
    let metadata = read_file_metadata(&mut file)?;
    let reader = FileReader::new(file, metadata, None);
    reader.collect::<Result<Vec<RecordBatch>>>()?;
    Ok(())
}
```

It produces:
```
Error: Io(Error { kind: UnexpectedEof, message: "failed to fill whole buffer" })
```
I'm running it on latest nightly but I don't think it makes any difference.


Pull request description:
Fixed error in compressing IPC LZ4
Closes #590. Thanks @dimasukr for the report!

Patch:
diff --git a/src/array/binary/mutable.rs b/src/array/binary/mutable.rs
index 28b1ebb08d..ae631ab7fd 100644
--- a/src/array/binary/mutable.rs
+++ b/src/array/binary/mutable.rs
@@ -92,6 +92,19 @@ impl<O: Offset> MutableBinaryArray<O> {
         }
     }
 
+    /// Initializes a new [`MutableBinaryArray`] with a pre-allocated capacity of slots and values.
+    pub fn with_capacities(capacity: usize, values: usize) -> Self {
+        let mut offsets = MutableBuffer::<O>::with_capacity(capacity + 1);
+        offsets.push(O::default());
+
+        Self {
+            data_type: Self::default_data_type(),
+            offsets,
+            values: MutableBuffer::<u8>::with_capacity(values),
+            validity: None,
+        }
+    }
+
     /// Reserves `additional` slots.
     pub fn reserve(&mut self, additional: usize) {
         self.offsets.reserve(additional);
@@ -145,6 +158,18 @@ impl<O: Offset> MutableBinaryArray<O> {
     }
 }
 
+impl<O: Offset> MutableBinaryArray<O> {
+    /// returns its values.
+    pub fn values(&self) -> &MutableBuffer<u8> {
+        &self.values
+    }
+
+    /// returns its offsets.
+    pub fn offsets(&self) -> &MutableBuffer<O> {
+        &self.offsets
+    }
+}
+
 impl<O: Offset> MutableArray for MutableBinaryArray<O> {
     fn len(&self) -> usize {
         self.offsets.len() - 1
@@ -228,15 +253,26 @@ impl<O: Offset> MutableBinaryArray<O> {
     }
 
     /// Creates a new [`BinaryArray`] from a [`TrustedLen`] of `&[u8]`.
+    /// # Safety
+    /// The iterator must be [`TrustedLen`](https://doc.rust-lang.org/std/iter/trait.TrustedLen.html).
+    /// I.e. that `size_hint().1` correctly reports its length.
     #[inline]
-    pub fn from_trusted_len_values_iter<T: AsRef<[u8]>, I: TrustedLen<Item = T>>(
+    pub unsafe fn from_trusted_len_values_iter_unchecked<T: AsRef<[u8]>, I: Iterator<Item = T>>(
         iterator: I,
     ) -> Self {
-        // soundness: I is `TrustedLen`
         let (offsets, values) = unsafe { trusted_len_values_iter(iterator) };
         Self::from_data(Self::default_data_type(), offsets, values, None)
     }
 
+    /// Creates a new [`BinaryArray`] from a [`TrustedLen`] of `&[u8]`.
+    #[inline]
+    pub fn from_trusted_len_values_iter<T: AsRef<[u8]>, I: TrustedLen<Item = T>>(
+        iterator: I,
+    ) -> Self {
+        // soundness: I is `TrustedLen`
+        unsafe { Self::from_trusted_len_values_iter_unchecked(iterator) }
+    }
+
     /// Creates a [`MutableBinaryArray`] from an falible iterator of trusted length.
     /// # Safety
     /// The iterator must be [`TrustedLen`](https://doc.rust-lang.org/std/iter/trait.TrustedLen.html).
@@ -252,7 +288,11 @@ impl<O: Offset> MutableBinaryArray<O> {
         let iterator = iterator.into_iter();
 
         // soundness: assumed trusted len
-        let (validity, offsets, values) = try_trusted_len_unzip(iterator)?;
+        let (mut validity, offsets, values) = try_trusted_len_unzip(iterator)?;
+
+        if validity.as_mut().unwrap().null_count() == 0 {
+            validity = None;
+        }
 
         Ok(Self::from_data(
             Self::default_data_type(),
diff --git a/src/io/ipc/compression.rs b/src/io/ipc/compression.rs
index a26a95b450..b98d670b79 100644
--- a/src/io/ipc/compression.rs
+++ b/src/io/ipc/compression.rs
@@ -33,7 +33,8 @@ pub fn decompress_zstd(_input_buf: &[u8], _output_buf: &mut [u8]) -> Result<()>
 pub fn compress_lz4(input_buf: &[u8], output_buf: &mut Vec<u8>) -> Result<()> {
     use std::io::Write;
     let mut encoder = lz4::EncoderBuilder::new().build(output_buf).unwrap();
-    encoder.write_all(input_buf).map_err(|e| e.into())
+    encoder.write_all(input_buf)?;
+    encoder.finish().1.map_err(|e| e.into())
 }
 
 #[cfg(feature = "io_ipc_compression")]
@@ -62,7 +63,7 @@ mod tests {
 
     #[cfg(feature = "io_ipc_compression")]
     #[test]
-    fn round_trip() {
+    fn round_trip_zstd() {
         let data: Vec<u8> = (0..200u8).map(|x| x % 10).collect();
         let mut buffer = vec![];
         compress_zstd(&data, &mut buffer).unwrap();
@@ -71,4 +72,16 @@ mod tests {
         decompress_zstd(&buffer, &mut result).unwrap();
         assert_eq!(data, result);
     }
+
+    #[cfg(feature = "io_ipc_compression")]
+    #[test]
+    fn round_trip_lz4() {
+        let data: Vec<u8> = (0..200u8).map(|x| x % 10).collect();
+        let mut buffer = vec![];
+        compress_lz4(&data, &mut buffer).unwrap();
+
+        let mut result = vec![0; 200];
+        decompress_lz4(&buffer, &mut result).unwrap();
+        assert_eq!(data, result);
+    }
 }


Test patch:
diff --git a/tests/it/array/binary/mutable.rs b/tests/it/array/binary/mutable.rs
index bbf7943144..7258dea133 100644
--- a/tests/it/array/binary/mutable.rs
+++ b/tests/it/array/binary/mutable.rs
@@ -1,5 +1,68 @@
-use arrow2::array::{BinaryArray, MutableBinaryArray};
+use std::ops::Deref;
+
+use arrow2::array::{BinaryArray, MutableArray, MutableBinaryArray};
 use arrow2::bitmap::Bitmap;
+use arrow2::error::ArrowError;
+
+#[test]
+fn new() {
+    assert_eq!(MutableBinaryArray::<i32>::new().len(), 0);
+
+    let a = MutableBinaryArray::<i32>::with_capacity(2);
+    assert_eq!(a.len(), 0);
+    assert!(a.offsets().capacity() >= 3);
+    assert_eq!(a.values().capacity(), 0);
+
+    let a = MutableBinaryArray::<i32>::with_capacities(2, 60);
+    assert_eq!(a.len(), 0);
+    assert!(a.offsets().capacity() >= 3);
+    assert!(a.values().capacity() >= 60);
+}
+
+#[test]
+fn from_iter() {
+    let iter = (0..3u8).map(|x| Some(vec![x; x as usize]));
+    let a: MutableBinaryArray<i32> = iter.clone().collect();
+    assert_eq!(a.values().deref(), &[1u8, 2, 2]);
+    assert_eq!(a.offsets().deref(), &[0, 0, 1, 3]);
+    assert_eq!(a.validity(), None);
+
+    let a = unsafe { MutableBinaryArray::<i32>::from_trusted_len_iter_unchecked(iter) };
+    assert_eq!(a.values().deref(), &[1u8, 2, 2]);
+    assert_eq!(a.offsets().deref(), &[0, 0, 1, 3]);
+    assert_eq!(a.validity(), None);
+}
+
+#[test]
+fn from_trusted_len_iter() {
+    let iter = (0..3u8).map(|x| vec![x; x as usize]);
+    let a: MutableBinaryArray<i32> = iter.clone().map(Some).collect();
+    assert_eq!(a.values().deref(), &[1u8, 2, 2]);
+    assert_eq!(a.offsets().deref(), &[0, 0, 1, 3]);
+    assert_eq!(a.validity(), None);
+
+    let a = unsafe {
+        MutableBinaryArray::<i32>::from_trusted_len_iter_unchecked(iter.clone().map(Some))
+    };
+    assert_eq!(a.values().deref(), &[1u8, 2, 2]);
+    assert_eq!(a.offsets().deref(), &[0, 0, 1, 3]);
+    assert_eq!(a.validity(), None);
+
+    let a = unsafe {
+        MutableBinaryArray::<i32>::try_from_trusted_len_iter_unchecked::<ArrowError, _, _>(
+            iter.clone().map(Some).map(Ok),
+        )
+    }
+    .unwrap();
+    assert_eq!(a.values().deref(), &[1u8, 2, 2]);
+    assert_eq!(a.offsets().deref(), &[0, 0, 1, 3]);
+    assert_eq!(a.validity(), None);
+
+    let a = unsafe { MutableBinaryArray::<i32>::from_trusted_len_values_iter_unchecked(iter) };
+    assert_eq!(a.values().deref(), &[1u8, 2, 2]);
+    assert_eq!(a.offsets().deref(), &[0, 0, 1, 3]);
+    assert_eq!(a.validity(), None);
+}
 
 #[test]
 fn push_null() {