refactor: bind scan expressions once

Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>

Author: joseph-isaacs

benchmarks/compress-bench/src/vortex.rs

@@ -18,6 +18,7 @@ use vortex::expr::root;
 use vortex::expr::select;
 use vortex::file::OpenOptionsSessionExt;
 use vortex::file::WriteOptionsSessionExt;
+use vortex::layout::scan::scan_builder::optimize_and_bind;
 use vortex_arrow::ToArrowType;
 use vortex_bench::Format;
 use vortex_bench::SESSION;
@@ -64,14 +65,15 @@ impl Compressor for VortexCompressor {
         let start = Instant::now();
         let data = Bytes::from(buf);
         let mut scan = SESSION.open_options().open_buffer(data)?.scan()?;
-        let root_columns = scan
-            .dtype()?
+        let source_dtype = scan.dtype()?;
+        let root_columns = source_dtype
             .as_struct_fields_opt()
             .map_or(0, |fields| fields.nfields());
         if let Some(cols) = read_projection(root_columns) {
             // Columns are named "0".."num_columns-1"; project the given subset.
             let names: FieldNames = cols.iter().map(|i| i.to_string()).collect();
-            scan = scan.with_projection(select(names, root()));
+            let projection = optimize_and_bind(select(names, root()), &source_dtype)?;
+            scan = scan.with_projection(projection);
         }
         let schema = Arc::new(scan.dtype()?.to_arrow_schema()?);
 

docs/developer-guide/internals/session.md

@@ -78,8 +78,9 @@ session.write_options()
     .await?;
 
 // Scanning a layout
+let filter = optimize_and_bind(expr, layout_reader.dtype())?;
 ScanBuilder::new(session.clone(), layout_reader)
-    .with_filter(expr)
+    .with_filter(filter)
     .into_array_stream()?;
 ```
 

fuzz/fuzz_targets/file_io.rs

@@ -6,6 +6,7 @@
 use itertools::Itertools;
 use libfuzzer_sys::Corpus;
 use libfuzzer_sys::fuzz_target;
+use vortex::layout::scan::scan_builder::optimize_and_bind;
 use vortex_array::Canonical;
 use vortex_array::IntoArray;
 use vortex_array::VortexSessionExecute;
@@ -78,14 +79,21 @@ fuzz_target!(|fuzz: FuzzFileAction| -> Corpus {
         .write(&mut full_buff, array_data.to_array_iterator())
         .vortex_expect("file write should succeed in fuzz test");
 
-    let mut output = SESSION
+    let file = SESSION
         .open_options()
         .open_buffer(full_buff)
-        .vortex_expect("open_buffer should succeed in fuzz test")
+        .vortex_expect("open_buffer should succeed in fuzz test");
+    let projection = optimize_and_bind(projection_expr.unwrap_or_else(root), file.dtype())
+        .vortex_expect("projection should bind in fuzz test");
+    let filter = filter_expr
+        .map(|filter| optimize_and_bind(filter, file.dtype()))
+        .transpose()
+        .vortex_expect("filter should bind in fuzz test");
+    let mut output = file
         .scan()
         .vortex_expect("scan should succeed in fuzz test")
-        .with_projection(projection_expr.unwrap_or_else(root))
-        .with_some_filter(filter_expr)
+        .with_projection(projection)
+        .with_some_filter(filter)
         .into_array_iter(&*RUNTIME)
         .vortex_expect("into_array_iter should succeed in fuzz test")
         .try_collect::<_, Vec<_>, _>()

vortex-bench/src/datasets/tpch_l_comment.rs

@@ -16,6 +16,7 @@ use vortex::dtype::Nullability::NonNullable;
 use vortex::expr::col;
 use vortex::expr::pack;
 use vortex::file::OpenOptionsSessionExt;
+use vortex::layout::scan::scan_builder::optimize_and_bind;
 
 use crate::Format;
 use crate::IdempotentPath;
@@ -66,9 +67,13 @@ impl Dataset for TPCHLCommentChunked {
 
         let path = data_dir.join("lineitem.vortex");
         let file = SESSION.open_options().open_path(path).await?;
+        let projection = optimize_and_bind(
+            pack(vec![("l_comment", col("l_comment"))], NonNullable),
+            file.dtype(),
+        )?;
         let chunks: Vec<_> = file
             .scan()?
-            .with_projection(pack(vec![("l_comment", col("l_comment"))], NonNullable))
+            .with_projection(projection)
             .map({
                 let ctx = ctx.clone();
                 move |a| {

vortex-datafusion/src/persistent/opener.rs

@@ -48,6 +48,7 @@ use vortex::file::OpenOptionsSessionExt;
 use vortex::io::InstrumentedReadAt;
 use vortex::layout::LayoutReader;
 use vortex::layout::scan::scan_builder::ScanBuilder;
+use vortex::layout::scan::scan_builder::optimize_and_bind;
 use vortex::layout::scan::split_by::SplitBy;
 use vortex::metrics::Label;
 use vortex::metrics::MetricsRegistry;
@@ -302,9 +303,11 @@ impl FileOpener for VortexOpener {
 
             // The schema of the stream returned from the vortex scan.
             // We use a reference schema for types that don't roundtrip (Dictionary, Utf8, etc.).
-            let scan_dtype = scan_projection.return_dtype(vxf.dtype()).map_err(|_e| {
-                exec_datafusion_err!("Couldn't get the dtype for the underlying Vortex scan")
-            })?;
+            let scan_projection =
+                optimize_and_bind(scan_projection, vxf.dtype()).map_err(|_e| {
+                    exec_datafusion_err!("Couldn't get the dtype for the underlying Vortex scan")
+                })?;
+            let scan_dtype = scan_projection.dtype().clone();
 
             // When projection pushdown is enabled, the scan outputs the projected columns.
             // When disabled, the scan outputs raw columns and the projection is applied after.
@@ -419,6 +422,10 @@ impl FileOpener for VortexOpener {
                     make_vortex_predicate(expr_convertor.as_ref(), &pushed).transpose()
                 })
                 .transpose()?;
+            let filter = filter
+                .map(|filter| optimize_and_bind(filter, vxf.dtype()))
+                .transpose()
+                .map_err(|e| exec_datafusion_err!("Couldn't bind Vortex scan filter: {e}"))?;
 
             if let Some(limit) = limit
                 && filter.is_none()