More Async IPC (#313)

Cargo.toml

@@ -95,6 +95,7 @@ tokio = "1.37.0"
 uninit = "0.6.2"
 uuid = "1.8.0"
 walkdir = "2.5.0"
+worker = "0.2.0"
 zigzag = "0.1.0"
 
 [workspace.lints.rust]

vortex-buffer/src/lib.rs

@@ -41,6 +41,13 @@ impl Buffer {
         }
     }
 
+    pub fn as_bytes(&self) -> &[u8] {
+        match self {
+            Buffer::Arrow(b) => b.as_ref(),
+            Buffer::Bytes(b) => b.as_ref(),
+        }
+    }
+
     pub fn into_vec(self) -> Result<Vec<u8>, Buffer> {
         match self {
             Buffer::Arrow(buffer) => buffer.into_vec::<u8>().map_err(Buffer::Arrow),
@@ -54,19 +61,13 @@ impl Deref for Buffer {
     type Target = [u8];
 
     fn deref(&self) -> &Self::Target {
-        match self {
-            Buffer::Arrow(b) => b.deref(),
-            Buffer::Bytes(b) => b.deref(),
-        }
+        self.as_bytes()
     }
 }
 
 impl AsRef<[u8]> for Buffer {
     fn as_ref(&self) -> &[u8] {
-        match self {
-            Buffer::Arrow(b) => b.as_ref(),
-            Buffer::Bytes(b) => b.as_ref(),
-        }
+        self.as_bytes()
     }
 }
 

vortex-error/Cargo.toml

@@ -21,6 +21,7 @@ flatbuffers = { workspace = true }
 flexbuffers = { workspace = true, optional = true }
 parquet = { workspace = true, optional = true }
 thiserror = { workspace = true }
+worker = { workspace = true, optional = true }
 
 [lints]
 workspace = true

vortex-error/src/lib.rs

@@ -120,6 +120,13 @@ pub enum VortexError {
         #[backtrace]
         std::array::TryFromSliceError,
     ),
+    #[cfg(feature = "worker")]
+    #[error(transparent)]
+    WorkerError(
+        #[from]
+        #[backtrace]
+        worker::Error,
+    ),
 }
 
 pub type VortexResult<T> = Result<T, VortexError>;
@@ -190,3 +197,10 @@ pub mod __private {
         error
     }
 }
+
+#[cfg(feature = "worker")]
+impl From<VortexError> for worker::Error {
+    fn from(value: VortexError) -> Self {
+        worker::Error::RustError(value.to_string())
+    }
+}

vortex-fastlanes/src/for/compress.rs

@@ -190,7 +190,7 @@ mod test {
             .compress(array.array(), None, Compressor::new(&ctx()))
             .unwrap();
         let compressed = FoRArray::try_from(compressed).unwrap();
-        assert_eq!(i8::MIN, compressed.reference().try_into().unwrap());
+        assert_eq!(i8::MIN, i8::try_from(compressed.reference()).unwrap());
 
         let encoded = compressed.encoded().flatten_primitive().unwrap();
         let bitcast: &[u8] = unsafe { std::mem::transmute(encoded.typed_data::<i8>()) };
@@ -206,12 +206,7 @@ mod test {
             .for_each(|(i, v)| {
                 assert_eq!(
                     *v,
-                    compressed
-                        .scalar_at(i)
-                        .unwrap()
-                        .as_ref()
-                        .try_into()
-                        .unwrap()
+                    i8::try_from(compressed.scalar_at(i).unwrap().as_ref()).unwrap()
                 );
             });
     }

vortex-ipc/Cargo.toml

@@ -22,7 +22,6 @@ log = { workspace = true }
 monoio = { workspace = true, optional = true, features = ["bytes"] }
 nougat = "0.2.4"
 pin-project = { workspace = true }
-tokio = { workspace = true, optional = true }
 vortex-array = { path = "../vortex-array" }
 vortex-buffer = { path = "../vortex-buffer" }
 vortex-error = { path = "../vortex-error" }
@@ -51,10 +50,9 @@ arrow-select = { workspace = true }
 workspace = true
 
 [features]
-default = ["futures", "monoio", "tokio"]
+default = ["futures", "monoio"]
 futures = []
 monoio = ["dep:monoio"]
-tokio = ["dep:tokio"]
 
 [[bench]]
 name = "ipc_take"