chore: Rework array data for mutability

crates/docgen/src/markdown_table.rs

@@ -1,7 +1,7 @@
 use std::fmt;
 
 use rayexec_error::Result;
-use rayexec_execution::arrays::batch::Batch;
+use rayexec_execution::arrays::batch::Batch2;
 use rayexec_execution::arrays::field::Schema;
 use rayexec_execution::arrays::format::{FormatOptions, Formatter};
 
@@ -13,7 +13,7 @@ const FORMATTER: Formatter = Formatter::new(FormatOptions {
 pub fn write_markdown_table<'a>(
     output: &mut dyn fmt::Write,
     schema: &Schema,
-    batches: impl IntoIterator<Item = &'a Batch>,
+    batches: impl IntoIterator<Item = &'a Batch2>,
 ) -> Result<()> {
     // 'field1 | field2 | field3'
     let header = schema
@@ -54,17 +54,17 @@ pub fn write_markdown_table<'a>(
 
 #[cfg(test)]
 mod tests {
-    use rayexec_execution::arrays::array::Array;
+    use rayexec_execution::arrays::array::Array2;
     use rayexec_execution::arrays::datatype::DataType;
     use rayexec_execution::arrays::field::Field;
 
     use super::*;
 
     #[test]
     fn simple() {
-        let batch = Batch::try_new([
-            Array::from_iter([1, 2, 3]),
-            Array::from_iter(["cat", "dog", "mouse"]),
+        let batch = Batch2::try_new([
+            Array2::from_iter([1, 2, 3]),
+            Array2::from_iter(["cat", "dog", "mouse"]),
         ])
         .unwrap();
 

crates/rayexec_csv/src/copy_to.rs

@@ -1,7 +1,7 @@
 use futures::future::BoxFuture;
 use futures::FutureExt;
 use rayexec_error::Result;
-use rayexec_execution::arrays::batch::Batch;
+use rayexec_execution::arrays::batch::Batch2;
 use rayexec_execution::arrays::field::Schema;
 use rayexec_execution::execution::operators::sink::PartitionSink;
 use rayexec_execution::functions::copy::CopyToFunction;
@@ -53,7 +53,7 @@ pub struct CsvCopyToSink {
 }
 
 impl CsvCopyToSink {
-    async fn push_inner(&mut self, batch: Batch) -> Result<()> {
+    async fn push_inner(&mut self, batch: Batch2) -> Result<()> {
         let mut buf = Vec::with_capacity(1024);
         self.encoder.encode(&batch, &mut buf)?;
         self.sink.write_all(buf.into()).await?;
@@ -68,7 +68,7 @@ impl CsvCopyToSink {
 }
 
 impl PartitionSink for CsvCopyToSink {
-    fn push(&mut self, batch: Batch) -> BoxFuture<'_, Result<()>> {
+    fn push(&mut self, batch: Batch2) -> BoxFuture<'_, Result<()>> {
         self.push_inner(batch).boxed()
     }
 

crates/rayexec_csv/src/datatable.rs

@@ -2,7 +2,7 @@ use std::fmt::{self, Debug};
 
 use futures::future::BoxFuture;
 use rayexec_error::Result;
-use rayexec_execution::arrays::batch::Batch;
+use rayexec_execution::arrays::batch::Batch2;
 use rayexec_execution::runtime::Runtime;
 use rayexec_execution::storage::table_storage::{
     DataTable,
@@ -60,7 +60,7 @@ pub struct CsvFileScan {
 }
 
 impl DataTableScan for CsvFileScan {
-    fn pull(&mut self) -> BoxFuture<'_, Result<Option<Batch>>> {
+    fn pull(&mut self) -> BoxFuture<'_, Result<Option<Batch2>>> {
         Box::pin(async { self.reader.read_next().await })
     }
 }

crates/rayexec_csv/src/reader.rs

@@ -23,8 +23,8 @@ use bytes::Bytes;
 use futures::stream::BoxStream;
 use futures::StreamExt;
 use rayexec_error::{RayexecError, Result};
-use rayexec_execution::arrays::array::{Array, ArrayData};
-use rayexec_execution::arrays::batch::Batch;
+use rayexec_execution::arrays::array::{Array2, ArrayData2};
+use rayexec_execution::arrays::batch::Batch2;
 use rayexec_execution::arrays::bitmap::Bitmap;
 use rayexec_execution::arrays::compute::cast::parse::{
     BoolParser,
@@ -342,7 +342,7 @@ impl AsyncCsvReader {
         AsyncCsvReader { stream }
     }
 
-    pub async fn read_next(&mut self) -> Result<Option<Batch>> {
+    pub async fn read_next(&mut self) -> Result<Option<Batch2>> {
         self.stream.next_batch().await
     }
 }
@@ -387,7 +387,7 @@ struct AsyncCsvStream {
 }
 
 impl AsyncCsvStream {
-    async fn next_batch(&mut self) -> Result<Option<Batch>> {
+    async fn next_batch(&mut self) -> Result<Option<Batch2>> {
         loop {
             let (buf, offset) = match self.buf.take() {
                 Some(buf) => (buf, self.buf_offset),
@@ -455,7 +455,7 @@ impl AsyncCsvStream {
         completed: CompletedRecords,
         schema: &Schema,
         skip_header: bool,
-    ) -> Result<Batch> {
+    ) -> Result<Batch2> {
         let skip_records = if skip_header { 1 } else { 0 };
 
         let mut arrs = Vec::with_capacity(schema.fields.len());
@@ -483,14 +483,14 @@ impl AsyncCsvStream {
             arrs.push(arr);
         }
 
-        Batch::try_new(arrs)
+        Batch2::try_new(arrs)
     }
 
     fn build_boolean(
         completed: &CompletedRecords,
         field_idx: usize,
         skip_records: usize,
-    ) -> Result<Array> {
+    ) -> Result<Array2> {
         let mut values = Bitmap::with_capacity(completed.num_completed());
         let mut validity = Bitmap::with_capacity(completed.num_completed());
 
@@ -507,7 +507,7 @@ impl AsyncCsvStream {
             }
         }
 
-        Ok(Array::new_with_validity_and_array_data(
+        Ok(Array2::new_with_validity_and_array_data(
             DataType::Boolean,
             validity,
             BooleanStorage::from(values),
@@ -520,11 +520,11 @@ impl AsyncCsvStream {
         field_idx: usize,
         skip_records: usize,
         mut parser: P,
-    ) -> Result<Array>
+    ) -> Result<Array2>
     where
         T: Default,
         P: Parser<Type = T>,
-        PrimitiveStorage<T>: Into<ArrayData>,
+        PrimitiveStorage<T>: Into<ArrayData2>,
     {
         let mut values = Vec::with_capacity(completed.num_completed());
         let mut validity = Bitmap::with_capacity(completed.num_completed());
@@ -544,7 +544,7 @@ impl AsyncCsvStream {
             }
         }
 
-        Ok(Array::new_with_validity_and_array_data(
+        Ok(Array2::new_with_validity_and_array_data(
             datatype.clone(),
             validity,
             PrimitiveStorage::from(values),
@@ -555,7 +555,7 @@ impl AsyncCsvStream {
         completed: &CompletedRecords,
         field_idx: usize,
         skip_records: usize,
-    ) -> Result<Array> {
+    ) -> Result<Array2> {
         let mut values = GermanVarlenBuffer::with_len(completed.num_completed() - skip_records);
         let mut validity = Bitmap::with_capacity(completed.num_completed());
 
@@ -569,7 +569,7 @@ impl AsyncCsvStream {
             }
         }
 
-        Ok(Array::new_with_validity_and_array_data(
+        Ok(Array2::new_with_validity_and_array_data(
             DataType::Utf8,
             validity,
             values.into_data(),

crates/rayexec_csv/src/writer.rs

@@ -2,7 +2,7 @@ use std::io::Write as _;
 
 use csv::ByteRecord;
 use rayexec_error::{Result, ResultExt};
-use rayexec_execution::arrays::batch::Batch;
+use rayexec_execution::arrays::batch::Batch2;
 use rayexec_execution::arrays::field::Schema;
 use rayexec_execution::arrays::format::{FormatOptions, Formatter};
 
@@ -38,7 +38,7 @@ impl CsvEncoder {
         }
     }
 
-    pub fn encode(&mut self, batch: &Batch, output_buf: &mut Vec<u8>) -> Result<()> {
+    pub fn encode(&mut self, batch: &Batch2, output_buf: &mut Vec<u8>) -> Result<()> {
         const FORMATTER: Formatter = Formatter::new(FormatOptions::new());
 
         let mut csv_writer = csv::WriterBuilder::new()

crates/rayexec_debug/src/discard.rs

@@ -1,6 +1,6 @@
 use futures::future::BoxFuture;
 use rayexec_error::Result;
-use rayexec_execution::arrays::batch::Batch;
+use rayexec_execution::arrays::batch::Batch2;
 use rayexec_execution::arrays::field::Schema;
 use rayexec_execution::execution::operators::sink::PartitionSink;
 use rayexec_execution::functions::copy::CopyToFunction;
@@ -33,7 +33,7 @@ impl CopyToFunction for DiscardCopyToFunction {
 struct DiscardCopyToSink;
 
 impl PartitionSink for DiscardCopyToSink {
-    fn push(&mut self, _batch: Batch) -> BoxFuture<'_, Result<()>> {
+    fn push(&mut self, _batch: Batch2) -> BoxFuture<'_, Result<()>> {
         Box::pin(async { Ok(()) })
     }
 

crates/rayexec_debug/src/table_storage.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use futures::future::BoxFuture;
 use parking_lot::Mutex;
 use rayexec_error::{RayexecError, Result};
-use rayexec_execution::arrays::batch::Batch;
+use rayexec_execution::arrays::batch::Batch2;
 use rayexec_execution::arrays::field::Field;
 use rayexec_execution::database::catalog_entry::CatalogEntry;
 use rayexec_execution::execution::operators::sink::PartitionSink;
@@ -28,7 +28,7 @@ pub struct TablePreload {
     pub schema: String,
     pub name: String,
     pub columns: Vec<Field>,
-    pub data: Batch,
+    pub data: Batch2,
 }
 
 #[derive(Debug, Default)]
@@ -124,7 +124,7 @@ impl TableStorage for DebugTableStorage {
 
 #[derive(Debug, Clone, Default)]
 pub struct DebugDataTable {
-    data: Arc<Mutex<Vec<Batch>>>,
+    data: Arc<Mutex<Vec<Batch2>>>,
 }
 
 impl DataTable for DebugDataTable {
@@ -168,23 +168,23 @@ impl DataTable for DebugDataTable {
 
 #[derive(Debug)]
 pub struct DebugDataTableScan {
-    data: Vec<Batch>,
+    data: Vec<Batch2>,
 }
 
 impl DataTableScan for DebugDataTableScan {
-    fn pull(&mut self) -> BoxFuture<'_, Result<Option<Batch>>> {
+    fn pull(&mut self) -> BoxFuture<'_, Result<Option<Batch2>>> {
         Box::pin(async { Ok(self.data.pop()) })
     }
 }
 
 #[derive(Debug)]
 pub struct DebugDataTableInsert {
-    collected: Vec<Batch>,
-    data: Arc<Mutex<Vec<Batch>>>,
+    collected: Vec<Batch2>,
+    data: Arc<Mutex<Vec<Batch2>>>,
 }
 
 impl PartitionSink for DebugDataTableInsert {
-    fn push(&mut self, batch: Batch) -> BoxFuture<'_, Result<()>> {
+    fn push(&mut self, batch: Batch2) -> BoxFuture<'_, Result<()>> {
         Box::pin(async {
             self.collected.push(batch);
             Ok(())

crates/rayexec_delta/src/datatable.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use futures::future::BoxFuture;
 use rayexec_error::Result;
-use rayexec_execution::arrays::batch::Batch;
+use rayexec_execution::arrays::batch::Batch2;
 use rayexec_execution::storage::table_storage::{DataTable, DataTableScan, Projections};
 
 use crate::protocol::table::{Table, TableScan};
@@ -34,7 +34,7 @@ struct DeltaTableScan {
 }
 
 impl DataTableScan for DeltaTableScan {
-    fn pull(&mut self) -> BoxFuture<'_, Result<Option<Batch>>> {
+    fn pull(&mut self) -> BoxFuture<'_, Result<Option<Batch2>>> {
         Box::pin(async { self.scan.read_next().await })
     }
 }

crates/rayexec_delta/src/protocol/table.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 
 use futures::{StreamExt, TryStreamExt};
 use rayexec_error::{not_implemented, RayexecError, Result, ResultExt};
-use rayexec_execution::arrays::batch::Batch;
+use rayexec_execution::arrays::batch::Batch2;
 use rayexec_execution::arrays::datatype::{DataType, DecimalTypeMeta, TimeUnit, TimestampTypeMeta};
 use rayexec_execution::arrays::field::{Field, Schema};
 use rayexec_execution::arrays::scalar::decimal::{Decimal128Type, DecimalType};
@@ -182,7 +182,7 @@ pub struct TableScan {
 
 impl TableScan {
     /// Read the next batch.
-    pub async fn read_next(&mut self) -> Result<Option<Batch>> {
+    pub async fn read_next(&mut self) -> Result<Option<Batch2>> {
         loop {
             if self.current.is_none() {
                 let path = match self.paths.pop_front() {

crates/rayexec_error/src/lib.rs

@@ -16,12 +16,10 @@ macro_rules! not_implemented {
 }
 
 // TODO: Implement partial eq on msg
-#[derive(Debug)]
 pub struct RayexecError {
     inner: Box<RayexecErrorInner>,
 }
 
-#[derive(Debug)]
 struct RayexecErrorInner {
     /// Message for the error.
     pub msg: String,
@@ -153,6 +151,14 @@ impl fmt::Display for RayexecError {
     }
 }
 
+impl fmt::Debug for RayexecError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // Just use the Display impl for Debug, significantly easier to read
+        // especially when the error contains a backtrace.
+        write!(f, "{self}")
+    }
+}
+
 impl Error for RayexecError {
     fn source(&self) -> Option<&(dyn Error + 'static)> {
         self.inner.source.as_ref().map(|e| e.as_ref() as _)

crates/rayexec_execution/Cargo.toml

@@ -10,6 +10,7 @@ rayexec_parser = { path = "../rayexec_parser" }
 # rayexec_bullet = { path = "../rayexec_bullet" }
 rayexec_io = { path = "../rayexec_io" }
 fmtutil = { path = "../fmtutil" }
+stdutil = { path = "../stdutil" }
 # stackutil = { path = "../stackutil" } TODO: psm hash issues when compiling to wasm on macos
 
 ahash = { workspace = true }