more experiments

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

Author: a10y

Cargo.lock

@@ -20,7 +20,6 @@ vortex-array = { workspace = true }
 vortex-buffer = { workspace = true }
 vortex-dtype = { workspace = true }
 vortex-error = { workspace = true }
-vortex-io = { workspace = true }
 vortex-mask = { workspace = true }
 vortex-proto = { workspace = true }
 vortex-scalar = { workspace = true }

encodings/sequence/Cargo.toml

@@ -65,7 +65,6 @@ vortex-compute = { workspace = true, default-features = true }
 vortex-dtype = { workspace = true, features = ["arrow", "serde"] }
 vortex-error = { workspace = true, features = ["prost"] }
 vortex-flatbuffers = { workspace = true, features = ["array"] }
-vortex-io = { workspace = true }
 vortex-mask = { workspace = true }
 vortex-metrics = { workspace = true }
 vortex-scalar = { workspace = true }

vortex-array/Cargo.toml

@@ -14,7 +14,7 @@ fn tokio_spawn(b: Bencher, work_ms: u64) {
     let handle = rt.handle().clone();
     b.bench_local(|| {
         // Spawn 1000 tasks that all do 1MS of CPU-blocking work.
-        let join_handles: Vec<_> = (0..1000)
+        let join_handles: Vec<_> = (0..100)
             .map(|_| handle.spawn(async move { thread::sleep(Duration::from_millis(work_ms)) }))
             .collect();
 
@@ -30,7 +30,7 @@ fn tokio_spawn_blocking(b: Bencher, work_ms: u64) {
     b.bench_local(|| {
         let work_ms = work_ms;
         // Spawn 1000 tasks that all do 1MS of CPU-blocking work.
-        let join_handles: Vec<_> = (0..1000)
+        let join_handles: Vec<_> = (0..100)
             .map(|_| handle.spawn_blocking(move || thread::sleep(Duration::from_millis(work_ms))))
             .collect();
 
@@ -46,7 +46,7 @@ fn tokio_unblock(b: Bencher, work_ms: u64) {
     b.bench_local(|| {
         // Spawn 1000 tasks that all do 1MS of CPU-blocking work.
         let work_ms = work_ms;
-        let join_handles: Vec<_> = (0..1000)
+        let join_handles: Vec<_> = (0..100)
             .map(|_| {
                 handle.spawn(blocking::unblock(move || {
                     thread::sleep(Duration::from_millis(work_ms.clone()))
@@ -65,7 +65,7 @@ fn smol_unblock(b: Bencher, work_ms: u64) {
     b.bench_local(|| {
         // Spawn 1000 tasks that all do 1MS of CPU-blocking work.
         let work_ms = work_ms;
-        let mut join_handles = Vec::with_capacity(1000);
+        let mut join_handles = Vec::with_capacity(100);
         exec.spawn_many(
             (0..1000).map(|_| {
                 blocking::unblock(move || thread::sleep(Duration::from_millis(work_ms.clone())))

vortex-io/benches/spawn.rs

@@ -3,6 +3,7 @@
 
 use std::sync::{Arc, LazyLock};
 
+use blocking::unblock;
 use futures::future::BoxFuture;
 use tracing::Instrument;
 
@@ -44,6 +45,7 @@ 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())
     }
 
     fn spawn_blocking(&self, task: Box<dyn FnOnce() + Send + 'static>) -> AbortHandleRef {
@@ -67,6 +69,7 @@ impl Executor for CurrentTokioRuntime {
         Box::new(
             tokio::runtime::Handle::current()
                 .spawn(async move { cpu() })
+                // .spawn(unblock(|| cpu()))
                 .abort_handle(),
         )
     }

vortex-io/src/runtime/tokio.rs

@@ -135,10 +135,17 @@ 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();
                     chunk
                         .statistics()
                         .compute_all(&Stat::all().collect::<Vec<_>>())?;
-                    Ok((sequence_id, compressor.compress_chunk(&chunk)?))
+                    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:?}");
+                    Ok((sequence_id, compress))
                 })
             })
             .buffered(self.concurrency);

vortex-layout/src/layouts/compressed.rs

@@ -55,10 +55,13 @@ vortex-zstd = { workspace = true, optional = true }
 [dev-dependencies]
 anyhow = { workspace = true }
 arrow-array = { workspace = true }
+console-subscriber = "0.5.0"
 divan = { workspace = true }
+futures = { workspace = true }
 itertools = { workspace = true }
 mimalloc = { workspace = true }
-parquet = { workspace = true }
+object_store = { workspace = true, features = ["http", "fs"] }
+parquet = { workspace = true, features = ["arrow", "async", "object_store"] }
 rand = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true, features = ["full"] }

vortex/Cargo.toml

@@ -0,0 +1,57 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::sync::Arc;
+
+use futures::stream::StreamExt;
+use object_store::http::HttpBuilder;
+use object_store::local::LocalFileSystem;
+use object_store::path::Path;
+use parquet::arrow::async_reader::ParquetObjectReader;
+use vortex::VortexSessionDefault;
+use vortex_array::ArrayRef;
+use vortex_array::arrow::FromArrowArray;
+use vortex_array::stream::ArrayStreamAdapter;
+use vortex_dtype::DType;
+use vortex_dtype::arrow::FromArrowType;
+use vortex_error::vortex_err;
+use vortex_file::WriteOptionsSessionExt;
+use vortex_session::VortexSession;
+
+#[tokio::main]
+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 = 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")),
+    )
+    .await?;
+
+    let stream = reader.build()?;
+
+    // Turn into a Vortex record batch stream
+    let vx_stream = ArrayStreamAdapter::new(
+        DType::from_arrow(stream.schema().as_ref()),
+        stream.map(|br| {
+            br.map_err(|e| vortex_err!("error: {e}"))
+                .map(|b| ArrayRef::from_arrow(b, false))
+        }),
+    );
+
+    // Write to file
+    let output = tokio::fs::File::create("/tmp/output.vortex").await?;
+    let session = VortexSession::default();
+    println!("begin writing");
+    let start = std::time::Instant::now();
+    session.write_options().write(output, vx_stream).await?;
+    let duration = start.elapsed();
+
+    println!("rewrote in {duration:?}");
+
+    Ok(())
+}