chore: rework LayoutBatchStream for legibility (#1245)

1. Expose `dtype` because all the callers want a dtype not a schema.
2. Avoid `vortex_expect` by storing known valid values in the
StreamingState.
3. Use `mem::take` on the state to allow (2) to work.

Author: danking

bench-vortex/benches/compress_noci.rs

@@ -134,7 +134,7 @@ fn vortex_decompress_read(runtime: &Runtime, buf: Arc<Vec<u8>>) -> VortexResult<
         );
 
         let stream = builder.build().await?;
-        let dtype = stream.schema().clone().into();
+        let dtype = stream.dtype().clone();
         let vecs: Vec<Array> = stream.try_collect().await?;
 
         ChunkedArray::try_new(vecs, dtype).map(|e| e.into())

vortex-serde/src/layouts/read/mask.rs

@@ -38,7 +38,12 @@ impl RowMask {
         Ok(Self { values, begin, end })
     }
 
-    /// Construct a RowMask from given bitmap and begin
+    /// Construct a RowMask which is valid in the given range.
+    pub fn new_valid_between(begin: usize, end: usize) -> Self {
+        unsafe { RowMask::new_unchecked(Bitmap::from_range(0..(end - begin) as u32), begin, end) }
+    }
+
+    /// Construct a RowMask from given bitmap and begin.
     ///
     /// # Safety
     ///

vortex-serde/src/layouts/read/recordbatchreader.rs

@@ -40,7 +40,7 @@ where
         stream: LayoutBatchStream<R>,
         runtime: &'a AR,
     ) -> VortexResult<VortexRecordBatchReader<'a, R, AR>> {
-        let arrow_schema = Arc::new(infer_schema(stream.schema().dtype())?);
+        let arrow_schema = Arc::new(infer_schema(stream.dtype())?);
         Ok(VortexRecordBatchReader {
             stream,
             arrow_schema,