Primitive and Bool array roundtrip serialization (#704)

vortex-array/src/view.rs

@@ -126,7 +126,8 @@ impl ArrayView {
             .iter()
             .take(idx)
             .map(|child| Self::cumulative_nbuffers(child))
-            .sum();
+            .sum::<usize>()
+            + self.has_buffer() as usize;
         let buffer_count = Self::cumulative_nbuffers(child);
 
         Some(Self {

vortex-serde/src/stream_writer/mod.rs

@@ -11,6 +11,9 @@ use vortex_error::VortexResult;
 use crate::io::VortexWrite;
 use crate::MessageWriter;
 
+#[cfg(test)]
+mod tests;
+
 pub struct StreamArrayWriter<W: VortexWrite> {
     msgs: MessageWriter<W>,
 

vortex-serde/src/stream_writer/tests.rs

@@ -0,0 +1,36 @@
+use std::io::Cursor;
+use std::sync::Arc;
+
+use arrow_array::cast::AsArray;
+use arrow_array::types::Int32Type;
+use arrow_array::PrimitiveArray;
+use vortex::arrow::FromArrowArray;
+use vortex::stream::ArrayStreamExt;
+use vortex::{Array, Context, IntoCanonical};
+
+use crate::stream_reader::StreamArrayReader;
+use crate::stream_writer::StreamArrayWriter;
+
+#[tokio::test]
+async fn broken_data() {
+    let arrow_arr: PrimitiveArray<Int32Type> = [Some(1), Some(2), Some(3), None].iter().collect();
+    let vortex_arr = Array::from_arrow(&arrow_arr, true);
+    let written = StreamArrayWriter::new(Vec::new())
+        .write_array(vortex_arr)
+        .await
+        .unwrap()
+        .into_inner();
+    let reader = StreamArrayReader::try_new(Cursor::new(written), Arc::new(Context::default()))
+        .await
+        .unwrap();
+    let arr = reader
+        .load_dtype()
+        .await
+        .unwrap()
+        .into_array_stream()
+        .collect_chunked()
+        .await
+        .unwrap();
+    let round_tripped = arr.into_canonical().unwrap().into_arrow();
+    assert_eq!(&arrow_arr, round_tripped.as_primitive::<Int32Type>());
+}