[wip] Runtimes + I/O

Cargo.toml

@@ -69,8 +69,11 @@ arrow-ord = "55.2.0"
 arrow-schema = "55.2.0"
 arrow-select = "55.2.0"
 arrow-string = "55.2.0"
+async-compat = "0.2.5"
 async-stream = "0.3.6"
+async-task = "4.7.1"
 async-trait = "0.1.88"
+boxcar = "0.2.14"
 bindgen = "0.72.0"
 bit-vec = "0.8.0"
 bitvec = "1.0.1"
@@ -82,6 +85,7 @@ cfg-if = "1"
 chrono = "0.4.41"
 clap = "4.5"
 compio = { version = "0.15", features = ["io-uring"], default-features = false }
+crossbeam-channel = "0.5.15"
 crossbeam-deque = "0.8.6"
 crossbeam-queue = "0.3.12"
 crossterm = "0.29"
@@ -102,7 +106,6 @@ enum-iterator = "2.0.0"
 erased-serde = "0.4"
 fastlanes = "0.2.0"
 flatbuffers = "25.2.10"
-flate2 = "1.1.2"
 fsst-rs = "0.5.2"
 futures = { version = "0.3.31", default-features = false }
 futures-executor = "0.3.31"
@@ -133,6 +136,7 @@ num-traits = "0.2.19"
 num_enum = { version = "0.7.3", default-features = false }
 object_store = { version = "0.12.3", default-features = false }
 once_cell = "1.21"
+oneshot = "0.1.11"
 opentelemetry = "0.30.0"
 opentelemetry-otlp = "0.30.0"
 opentelemetry_sdk = "0.30.0"
@@ -172,6 +176,7 @@ simdutf8 = "0.1.5"
 similar = "2.7.0"
 simplelog = "0.12"
 sketches-ddsketch = "0.3.0"
+smol = "2.0.2"
 static_assertions = "1.1"
 strum = "0.27"
 sysinfo = "0.37.0"

bench-vortex/src/compress/vortex.rs

@@ -5,16 +5,17 @@ use std::io::Cursor;
 use std::sync::Arc;
 
 use bytes::Bytes;
-use tokio::runtime::Handle;
-use vortex::Array;
+use tokio::runtime;
 use vortex::file::{VortexOpenOptions, VortexWriteOptions, WriteStrategyBuilder};
+use vortex::io::runtime::Handle;
+use vortex::Array;
 
 #[inline(never)]
 pub async fn vortex_compress_write(array: &dyn Array, buf: &mut Vec<u8>) -> anyhow::Result<u64> {
     Ok(VortexWriteOptions::default()
         .with_strategy(
             WriteStrategyBuilder::new()
-                .with_executor(Arc::new(Handle::current()))
+                .with_executor(Arc::new(runtime::Handle::current()))
                 .build(),
         )
         .write(Cursor::new(buf), array.to_array_stream())
@@ -24,7 +25,8 @@ pub async fn vortex_compress_write(array: &dyn Array, buf: &mut Vec<u8>) -> anyh
 
 #[inline(never)]
 pub async fn vortex_decompress_read(buf: Bytes) -> anyhow::Result<usize> {
-    let scan = VortexOpenOptions::in_memory().open(buf)?.scan()?;
+    let vxf = VortexOpenOptions::in_memory().open(buf, Handle::no_op())?;
+    let scan = vxf.scan()?;
     let schema = Arc::new(scan.dtype()?.to_arrow_schema()?);
 
     let iter = scan.into_record_batch_reader_multithread(schema)?;