more

Signed-off-by: Andrew Duffy <[email protected]>

Author: a10y

vortex-io/src/runtime/tokio.rs

@@ -44,8 +44,8 @@ impl Executor for tokio::runtime::Handle {
     }
 
     fn spawn_cpu(&self, cpu: Box<dyn FnOnce() + Send + 'static>) -> AbortHandleRef {
-        Box::new(tokio::runtime::Handle::spawn(self, async move { cpu() }).abort_handle())
-        // Box::new(tokio::runtime::Handle::spawn(self, unblock(|| cpu())).abort_handle())
+        // Box::new(tokio::runtime::Handle::spawn(self, async move { cpu() }).abort_handle())
+        Box::new(tokio::runtime::Handle::spawn(self, unblock(|| cpu())).abort_handle())
     }
 
     fn spawn_blocking(&self, task: Box<dyn FnOnce() + Send + 'static>) -> AbortHandleRef {
@@ -68,8 +68,8 @@ impl Executor for CurrentTokioRuntime {
     fn spawn_cpu(&self, cpu: Box<dyn FnOnce() + Send + 'static>) -> AbortHandleRef {
         Box::new(
             tokio::runtime::Handle::current()
-                .spawn(async move { cpu() })
-                // .spawn(unblock(|| cpu()))
+                // .spawn(async move { cpu() })
+                .spawn(unblock(|| cpu()))
                 .abort_handle(),
         )
     }

vortex-layout/src/layouts/compressed.rs

@@ -135,16 +135,16 @@ impl LayoutStrategy for CompressingStrategy {
                 handle2.spawn_cpu(move || {
                     let (sequence_id, chunk) = chunk?;
                     // Compute the stats for the chunk prior to compression
-                    let start = std::time::Instant::now();
+                    // let start = std::time::Instant::now();
                     chunk
                         .statistics()
                         .compute_all(&Stat::all().collect::<Vec<_>>())?;
-                    let elapsed = start.elapsed();
-                    println!("COMPUTE_ALL: {elapsed:?}");
-                    let start = std::time::Instant::now();
+                    // let elapsed = start.elapsed();
+                    // println!("COMPUTE_ALL: {elapsed:?}");
+                    // let start = std::time::Instant::now();
                     let compress = compressor.compress_chunk(&chunk)?;
-                    let elapsed = start.elapsed();
-                    println!("COMPRESS_CHUNK: {elapsed:?}");
+                    // let elapsed = start.elapsed();
+                    // println!("COMPRESS_CHUNK: {elapsed:?}");
                     Ok((sequence_id, compress))
                 })
             })

vortex/examples/rebuild_clickbench.rs

@@ -18,22 +18,24 @@ use vortex_error::vortex_err;
 use vortex_file::WriteOptionsSessionExt;
 use vortex_session::VortexSession;
 
-#[tokio::main]
+#[tokio::main(worker_threads = 8)]
 pub async fn main() -> anyhow::Result<()> {
     // console_subscriber::init();
-    // let store = LocalFileSystem::new();
-    let store = HttpBuilder::new()
-        .with_url("https://vortex-benchmark-results-database.s3.amazonaws.com")
-        .build()?;
+    let store = LocalFileSystem::new();
+    // let store = HttpBuilder::new()
+    //     .with_url("https://vortex-benchmark-results-database.s3.amazonaws.com")
+    //     .build()?;
     let store = Arc::new(store);
     let reader = parquet::arrow::async_reader::ParquetRecordBatchStreamBuilder::new(
-        ParquetObjectReader::new(store, Path::from("testing/hits.parquet")),
-        // ParquetObjectReader::new(store, Path::from("/Users/aduffy/Downloads/hits.parquet")),
+        // ParquetObjectReader::new(store, Path::from("testing/hits_83.parquet")),
+        ParquetObjectReader::new(store, Path::from("/Volumes/Code/vortex/bench-vortex/data/clickbench_partitioned/parquet/hits_83.parquet")),
     )
     .await?;
 
     let stream = reader.build()?;
 
+    // Start reading the next batch while we're recompressing the previous one.
+
     // Turn into a Vortex record batch stream
     let vx_stream = ArrayStreamAdapter::new(
         DType::from_arrow(stream.schema().as_ref()),