Support dynamic layouts with io batching

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)
     }

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::layout::writer::layout_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?;
 

vortex-datafusion/src/persistent/opener.rs

@@ -7,10 +7,11 @@ use datafusion_common::Result as DFResult;
 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::{Context, IntoCanonical};
 use vortex_serde::io::ObjectStoreReadAt;
+use vortex_serde::layout::reader::projections::Projection;
+use vortex_serde::layout::reader::VortexLayoutReaderBuilder;
+use vortex_serde::layout::{LayoutContext, LayoutReader};
 
 pub struct VortexFileOpener {
     pub object_store: Arc<dyn ObjectStore>,
@@ -24,7 +25,13 @@ 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,
+            LayoutReader::new(
+                Arc::new(Context::default()),
+                Arc::new(LayoutContext::default()),
+            ),
+        );
 
         if let Some(batch_size) = self.batch_size {
             builder = builder.with_batch_size(batch_size);

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 }