chore: rework LayoutBatchStream for legibility

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

@@ -36,7 +36,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
     ///
@@ -158,12 +163,12 @@ impl RowMask {
             return Ok(Some(sliced.clone()));
         }
 
-        let predicate = self.to_predicate_array()?;
+        let predicate = self.to_array()?;
 
         filter(sliced, predicate).map(Some)
     }
 
-    pub fn to_predicate_array(&self) -> VortexResult<Array> {
+    pub fn to_array(&self) -> VortexResult<Array> {
         let bitset = self
             .values
             .to_bitset()

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,