[PERF] native streaming parquet (#1193)

* Improves perf of parquet reader by performing pipelining between
compute and IO
* Uses rayon and oneshot channels to fan out expensive work to blocking
threads

Cargo.toml

@@ -51,6 +51,7 @@ members = [
 ]
 
 [workspace.dependencies]
+bytes = "1.4.0"
 futures = "0.3.28"
 html-escape = "0.2.13"
 num-derive = "0.3.3"

src/daft-io/Cargo.toml

@@ -6,7 +6,7 @@ aws-credential-types = {version = "0.55.3", features = ["hardcoded-credentials"]
 aws-sdk-s3 = "0.28.0"
 aws-sig-auth = "0.55.3"
 aws-sigv4 = "0.55.3"
-bytes = "1.4.0"
+bytes = {workspace = true}
 common-error = {path = "../common/error", default-features = false}
 daft-core = {path = "../daft-core", default-features = false}
 futures = {workspace = true}

src/daft-io/src/lib.rs

@@ -86,6 +86,12 @@ impl From<Error> for DaftError {
     }
 }
 
+impl From<Error> for std::io::Error {
+    fn from(err: Error) -> std::io::Error {
+        std::io::Error::new(std::io::ErrorKind::Other, err)
+    }
+}
+
 type Result<T, E = Error> = std::result::Result<T, E>;
 
 #[derive(Default)]

src/daft-parquet/Cargo.toml

@@ -1,5 +1,8 @@
 [dependencies]
 arrow2 = {workspace = true, features = ["io_parquet", "io_parquet_compression"]}
+async-compat = "0.2.1"
+async-stream = "0.3.5"
+bytes = {workspace = true}
 common-error = {path = "../common/error", default-features = false}
 daft-core = {path = "../daft-core", default-features = false}
 daft-io = {path = "../daft-io", default-features = false}
@@ -9,8 +12,11 @@ log = {workspace = true}
 parquet2 = "0.17.2"
 pyo3 = {workspace = true, optional = true}
 pyo3-log = {workspace = true, optional = true}
+rayon = "1.7.0"
 snafu = {workspace = true}
 tokio = {workspace = true}
+tokio-stream = "0.1.14"
+tokio-util = "0.7.8"
 
 [features]
 default = ["python"]