Support dynamic layouts with io batching (#533)

Allow for interleaving of on disk layouts
spiraldb · Aug 7, 2024 · 0f84596 · 0f84596
1 parent 7f57019
commit 0f84596
Show file tree

Hide file tree

Showing 34 changed files with 1,299 additions and 961 deletions.
diff --git a/.gitmodules b/.gitmodules
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/vortex-array/src/array/primitive/compute/mod.rs b/vortex-array/src/array/primitive/compute/mod.rs
@@ -25,6 +25,7 @@ impl ArrayCompute for PrimitiveArray {
     fn fill_forward(&self) -> Option<&dyn FillForwardFn> {
         Some(self)
     }
+
     fn filter_indices(&self) -> Option<&dyn FilterIndicesFn> {
         Some(self)
     }

diff --git a/vortex-datafusion/examples/table_provider.rs b/vortex-datafusion/examples/table_provider.rs
@@ -14,7 +14,7 @@ use vortex::validity::Validity;
 use vortex::IntoArray;
 use vortex_datafusion::persistent::config::{VortexFile, VortexTableConfig};
 use vortex_datafusion::persistent::provider::VortexFileTableProvider;
-use vortex_serde::file::file_writer::FileWriter;
+use vortex_serde::layouts::writer::LayoutWriter;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -48,7 +48,7 @@ async fn main() -> anyhow::Result<()> {
         .open(&filepath)
         .await?;
 
-    let writer = FileWriter::new(f);
+    let writer = LayoutWriter::new(f);
     let writer = writer.write_array_columns(st.into_array()).await?;
     writer.finalize().await?;
 

diff --git a/vortex-datafusion/src/persistent/opener.rs b/vortex-datafusion/src/persistent/opener.rs
@@ -8,9 +8,10 @@ use datafusion_physical_expr::PhysicalExpr;
 use futures::{FutureExt as _, TryStreamExt};
 use object_store::ObjectStore;
 use vortex::IntoCanonical;
-use vortex_serde::file::reader::projections::Projection;
-use vortex_serde::file::reader::VortexBatchReaderBuilder;
 use vortex_serde::io::ObjectStoreReadAt;
+use vortex_serde::layouts::reader::builder::VortexLayoutReaderBuilder;
+use vortex_serde::layouts::reader::context::LayoutDeserializer;
+use vortex_serde::layouts::reader::projections::Projection;
 
 pub struct VortexFileOpener {
     pub object_store: Arc<dyn ObjectStore>,
@@ -24,7 +25,7 @@ impl FileOpener for VortexFileOpener {
         let read_at =
             ObjectStoreReadAt::new(self.object_store.clone(), file_meta.location().clone());
 
-        let mut builder = VortexBatchReaderBuilder::new(read_at);
+        let mut builder = VortexLayoutReaderBuilder::new(read_at, LayoutDeserializer::default());
 
         if let Some(batch_size) = self.batch_size {
             builder = builder.with_batch_size(batch_size);

diff --git a/vortex-serde/Cargo.toml b/vortex-serde/Cargo.toml
@@ -12,6 +12,7 @@ edition = { workspace = true }
 rust-version = { workspace = true }
 
 [dependencies]
+ahash = { workspace = true }
 bytes = { workspace = true }
 flatbuffers = { workspace = true }
 futures = { workspace = true }
@@ -25,6 +26,7 @@ vortex-array = { workspace = true }
 vortex-buffer = { workspace = true }
 vortex-dtype = { workspace = true }
 vortex-error = { workspace = true, features = ["object_store"] }
+vortex-expr = { workspace = true }
 vortex-flatbuffers = { workspace = true }
 vortex-scalar = { workspace = true }