Use vortex-buffer over bytes::Bytes

Cargo.toml

@@ -46,14 +46,14 @@ categories = ["database-implementations", "data-structures", "compression"]
 
 [workspace.dependencies]
 anyhow = "1.0"
+aligned-buffer = "0.2.0"
 arbitrary = "1.3.2"
 arrayref = "0.3.7"
 arrow = { version = "53.0.0" }
 arrow-arith = "53.0.0"
 arrow-array = "53.0.0"
 arrow-buffer = "53.0.0"
 arrow-cast = "53.0.0"
-arrow-ipc = "53.0.0"
 arrow-ord = "53.0.0"
 arrow-schema = "53.0.0"
 arrow-select = "53.0.0"
@@ -103,6 +103,7 @@ once_cell = "1.20.2"
 parquet = "53.0.0"
 paste = "1.0.14"
 pin-project = "1.1.5"
+pin-project-lite = "0.2.15"
 prettytable-rs = "0.10.0"
 tabled = { version = "0.17.0", default-features = false }
 prost = "0.13.0"

bench-vortex/benches/bytes_at.rs

@@ -4,16 +4,12 @@
 use std::sync::Arc;
 
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
-use futures::executor::block_on;
-use futures::StreamExt;
 use vortex::array::{PrimitiveArray, VarBinArray, VarBinViewArray};
-use vortex::buffer::Buffer;
 use vortex::dtype::{DType, Nullability};
-use vortex::io::VortexBufReader;
-use vortex::ipc::stream_reader::StreamArrayReader;
-use vortex::ipc::stream_writer::StreamArrayWriter;
+use vortex::ipc::iterator::{ArrayIteratorIPC, SyncIPCReader};
+use vortex::iter::ArrayIteratorExt;
 use vortex::validity::Validity;
-use vortex::{Context, IntoArrayData, IntoCanonical};
+use vortex::{Context, IntoArrayData, IntoArrayVariant};
 
 fn array_data_fixture() -> VarBinArray {
     VarBinArray::try_new(
@@ -27,27 +23,16 @@ fn array_data_fixture() -> VarBinArray {
 
 fn array_view_fixture() -> VarBinViewArray {
     let array_data = array_data_fixture();
-    let mut buffer = Vec::new();
 
-    let writer = StreamArrayWriter::new(&mut buffer);
-    block_on(writer.write_array(array_data.into_array())).unwrap();
+    let buffer = array_data
+        .into_array()
+        .into_array_iterator()
+        .write_ipc(vec![])
+        .unwrap();
 
-    let buffer = Buffer::from(buffer);
-
-    let ctx = Arc::new(Context::default());
-    let reader = block_on(StreamArrayReader::try_new(
-        VortexBufReader::new(buffer),
-        ctx.clone(),
-    ))
-    .unwrap();
-    let reader = block_on(reader.load_dtype()).unwrap();
-
-    let mut stream = Box::pin(reader.into_array_stream());
-
-    block_on(stream.next())
-        .unwrap()
+    SyncIPCReader::try_new(buffer.as_slice(), Arc::new(Context::default()))
         .unwrap()
-        .into_canonical()
+        .into_array_data()
         .unwrap()
         .into_varbinview()
         .unwrap()

bench-vortex/src/data_downloads.rs

@@ -6,15 +6,16 @@ use std::path::PathBuf;
 
 use arrow_array::RecordBatchReader;
 use bzip2::read::BzDecoder;
+use futures::StreamExt;
 use log::info;
 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
 use tokio::runtime::Runtime;
 use vortex::array::ChunkedArray;
 use vortex::arrow::FromArrowType;
 use vortex::dtype::DType;
 use vortex::error::{VortexError, VortexResult};
-use vortex::io::TokioAdapter;
-use vortex::ipc::stream_writer::StreamArrayWriter;
+use vortex::io::{TokioAdapter, VortexWrite};
+use vortex::ipc::stream::ArrayStreamIPC;
 use vortex::{ArrayData, IntoArrayData};
 
 use crate::idempotent;
@@ -56,11 +57,11 @@ pub fn data_vortex_uncompressed(fname_out: &str, downloaded_data: PathBuf) -> Pa
         Runtime::new()
             .unwrap()
             .block_on(async move {
-                let write = TokioAdapter(tokio::fs::File::create(path).await.unwrap());
-                StreamArrayWriter::new(write)
-                    .write_array(array)
-                    .await
-                    .unwrap();
+                let mut write = TokioAdapter(tokio::fs::File::create(path).await.unwrap());
+                let mut bytes = array.into_array_stream().into_ipc();
+                while let Some(buffer) = bytes.next().await {
+                    write.write_all(buffer.unwrap()).await.unwrap();
+                }
                 Ok::<(), VortexError>(())
             })
             .unwrap();

vortex-array/Cargo.toml

@@ -32,7 +32,6 @@ arrow-schema = { workspace = true }
 arrow-select = { workspace = true }
 arrow-string = { workspace = true }
 backtrace = { workspace = true }
-bytes = { workspace = true }
 enum-iterator = { workspace = true }
 flatbuffers = { workspace = true }
 flexbuffers = { workspace = true }