[PERF] Add a parallel local CSV reader

Makefile

@@ -52,6 +52,10 @@ build: check-toolchain .venv  ## Compile and install Daft for development
 build-release: check-toolchain .venv  ## Compile and install a faster Daft binary
 	@unset CONDA_PREFIX && PYO3_PYTHON=$(VENV_BIN)/python $(VENV_BIN)/maturin develop --release
 
+.PHONY: build-bench
+build-bench: check-toolchain .venv  ## Compile and install a faster Daft binary
+	@unset CONDA_PREFIX && PYO3_PYTHON=$(VENV_BIN)/python $(VENV_BIN)/maturin develop --profile dev-bench
+
 .PHONY: test
 test: .venv build  ## Run tests
 	HYPOTHESIS_MAX_EXAMPLES=$(HYPOTHESIS_MAX_EXAMPLES) $(VENV_BIN)/pytest --hypothesis-seed=$(HYPOTHESIS_SEED)

src/arrow2/src/io/csv/read_async/reader.rs

@@ -1,10 +1,11 @@
 use futures::AsyncRead;
 
 use super::{AsyncReader, ByteRecord};
+use crate::io::csv::read;
 
 use crate::error::{Error, Result};
 
-/// Asynchronosly read `len` rows from `reader` into `row`, skipping the first `skip`.
+/// Asynchronosly read `rows.len` rows from `reader` into `rows`, skipping the first `skip`.
 /// This operation has minimal CPU work and is thus the fastest way to read through a CSV
 /// without deserializing the contents to Arrow.
 pub async fn read_rows<R>(
@@ -37,3 +38,25 @@ where
     }
     Ok(row_number)
 }
+
+/// Synchronously read `rows.len` rows from `reader` into `rows`. This is used in the local i/o case.
+pub fn local_read_rows<R>(
+    reader: &mut read::Reader<R>,
+    rows: &mut [read::ByteRecord],
+) -> Result<(usize, bool)>
+where
+    R: std::io::Read,
+{
+    let mut row_number = 0;
+    let mut has_more = true;
+    for row in rows.iter_mut() {
+        has_more = reader
+            .read_byte_record(row)
+            .map_err(|e| Error::External(format!(" at line {}", row_number), Box::new(e)))?;
+        if !has_more {
+            break;
+        }
+        row_number += 1;
+    }
+    Ok((row_number, has_more))
+}

src/daft-csv/Cargo.toml

@@ -5,6 +5,7 @@ async-compression = {workspace = true}
 async-stream = {workspace = true}
 common-error = {path = "../common/error", default-features = false}
 common-py-serde = {path = "../common/py-serde", default-features = false}
+crossbeam-channel = "0.5.1"
 csv-async = "1.3.0"
 daft-compression = {path = "../daft-compression", default-features = false}
 daft-core = {path = "../daft-core", default-features = false}
@@ -13,6 +14,9 @@ daft-dsl = {path = "../daft-dsl", default-features = false}
 daft-io = {path = "../daft-io", default-features = false}
 daft-table = {path = "../daft-table", default-features = false}
 futures = {workspace = true}
+indexmap = {workspace = true, features = ["serde"]}
+memchr = "2.7.2"
+memmap2 = "0.9.4"
 pyo3 = {workspace = true, optional = true}
 rayon = {workspace = true}
 serde = {workspace = true}

src/daft-csv/src/lib.rs

@@ -2,6 +2,8 @@
 #![feature(let_chains)]
 #![feature(trait_alias)]
 #![feature(trait_upcasting)]
+#![feature(test)]
+extern crate test;
 use common_error::DaftError;
 use snafu::Snafu;
 
@@ -23,6 +25,8 @@ pub enum Error {
     #[snafu(display("{source}"))]
     IOError { source: daft_io::Error },
     #[snafu(display("{source}"))]
+    StdIOError { source: std::io::Error },
+    #[snafu(display("{source}"))]
     CSVError { source: csv_async::Error },
     #[snafu(display("Invalid char: {}", val))]
     WrongChar {