[PERF] Add a parallel local CSV reader

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

@@ -4,6 +4,7 @@ async-compat = {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}
@@ -12,6 +13,7 @@ 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}
+memchr = "2.7.2"
 pyo3 = {workspace = true, optional = true}
 rayon = {workspace = true}
 serde = {workspace = true}

src/daft-csv/src/lib.rs

@@ -1,10 +1,14 @@
 #![feature(async_closure)]
 #![feature(let_chains)]
+#![feature(new_uninit)]
 #![feature(trait_alias)]
 #![feature(trait_upcasting)]
+#![feature(test)]
+extern crate test;
 use common_error::DaftError;
 use snafu::Snafu;
 
+pub mod local;
 pub mod metadata;
 pub mod options;
 #[cfg(feature = "python")]
@@ -23,6 +27,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 {