Optimize scanned arrays before the scan returns them

Layout readers apply the projection lazily, and apply no longer optimizes,
so a scan of a struct returned a pack expression rather than a struct.
vx_array_get_field, and the C++ Array::field built on it, require a struct
and rejected it. The scan now optimizes each split's array with its
session, as does the physical plan's eval stage.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SWbErbGQPZvkCK9ZXH87w
Signed-off-by: Claude <noreply@anthropic.com>

Author: claude

vortex-ffi/src/scan.rs

@@ -448,6 +448,7 @@ mod tests {
     use std::ptr;
     use std::sync::LazyLock;
 
+    use tempfile::NamedTempFile;
     use vortex::VortexSessionDefault;
     use vortex::array::IntoArray;
     use vortex::array::arrays::PrimitiveArray;
@@ -462,9 +463,12 @@ mod tests {
 
     use crate::array::vx_array;
     use crate::array::vx_array_free;
+    use crate::array::vx_array_get_field;
     use crate::data_source::vx_data_source_free;
     use crate::data_source::vx_data_source_new;
     use crate::data_source::vx_data_source_options;
+    use crate::dtype::vx_dtype;
+    use crate::dtype::vx_dtype_free;
     use crate::expression::vx_binary_operator;
     use crate::expression::vx_expression_binary;
     use crate::expression::vx_expression_free;
@@ -486,6 +490,9 @@ mod tests {
     use crate::scan::vx_scan_selection_include;
     use crate::session::vx_session_free;
     use crate::session::vx_session_new;
+    use crate::sink::vx_array_sink_close;
+    use crate::sink::vx_array_sink_open_file;
+    use crate::sink::vx_array_sink_push;
     use crate::string::vx_view;
     use crate::tests::SAMPLE_ROWS;
     use crate::tests::assert_no_error;
@@ -579,6 +586,62 @@ mod tests {
         unsafe { vx_array_free(array) };
     }
 
+    /// C callers read fields with `vx_array_get_field`, which needs a struct rather than the
+    /// lazy pack expression a layout reader builds for the projection.
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn test_scan_returns_struct() {
+        let mut ctx = SESSION.create_execution_ctx();
+        let strings = VarBinViewArray::from_iter_nullable_str([Some("inlined"), Some("y")]);
+        let struct_array = StructArray::try_new(
+            ["s"].into(),
+            vec![strings.clone().into_array()],
+            strings.len(),
+            Validity::NonNullable,
+        )
+        .unwrap();
+        unsafe {
+            let session = vx_session_new();
+            let file = NamedTempFile::new().unwrap();
+            let path = vx_view::from_str(file.path().to_str().unwrap());
+            let dtype = vx_dtype::new(struct_array.dtype().clone());
+            let mut error = ptr::null_mut();
+            let sink = vx_array_sink_open_file(session, path, dtype, &raw mut error);
+            assert_no_error(error);
+            let input = vx_array::new(struct_array.into_array());
+            vx_array_sink_push(sink, input, &raw mut error);
+            assert_no_error(error);
+            vx_array_sink_close(sink, &raw mut error);
+            assert_no_error(error);
+            vx_array_free(input);
+            vx_dtype_free(dtype);
+
+            let ds_options = vx_data_source_options {
+                paths: &raw const path,
+                paths_len: 1,
+            };
+            let ds = vx_data_source_new(session, &raw const ds_options, &raw mut error);
+            assert_no_error(error);
+            let scan = vx_data_source_scan(ds, ptr::null(), ptr::null_mut(), &raw mut error);
+            assert_no_error(error);
+            let partition = vx_scan_next_partition(scan, &raw mut error);
+            assert_no_error(error);
+            let array = vx_partition_next(partition, &raw mut error);
+            assert_no_error(error);
+
+            let field = vx_array_get_field(array, 0, &raw mut error);
+            assert_no_error(error);
+            assert_arrays_eq!(vx_array::as_ref(field), strings, &mut ctx);
+
+            vx_array_free(field);
+            vx_array_free(array);
+            vx_partition_free(partition);
+            vx_scan_free(scan);
+            vx_data_source_free(ds);
+            vx_session_free(session);
+        }
+    }
+
     #[test]
     #[cfg_attr(miri, ignore)]
     fn test_project_single_field() {

vortex-layout/src/plan/physical/operators/eval.rs

@@ -4,6 +4,7 @@
 //! Applying an expression to each batch.
 
 use vortex_array::expr::BoundExpression;
+use vortex_array::optimizer::ArrayOptimizer;
 use vortex_error::VortexResult;
 
 use crate::plan::physical::operator::Cx;
@@ -23,10 +24,13 @@ impl EvalStage {
 }
 
 impl Operator for EvalStage {
-    fn compute(&mut self, input: Input, _cx: &mut Cx<'_>) -> VortexResult<Step> {
+    fn compute(&mut self, input: Input, cx: &mut Cx<'_>) -> VortexResult<Step> {
         let Input::Batch(batch) = input else {
             return Ok(Step::End);
         };
-        Ok(Step::Final(batch.apply_bound(&self.expression)?))
+        // `apply_bound` builds the expression lazily. Optimizing it here hands consumers the
+        // reduced array, such as a struct for a pack of fields, rather than the expression node.
+        let applied = batch.apply_bound(&self.expression)?;
+        Ok(Step::Final(applied.optimize(cx.session())?))
     }
 }

vortex-layout/src/scan/repeated_scan.rs

@@ -165,6 +165,7 @@ impl<A: 'static + Send> RepeatedScan<A> {
                 .map(|filter| Arc::new(FilterExpr::new(filter))),
             reader: Arc::clone(&self.layout_reader),
             projection: self.projection.clone(),
+            session: self.session.clone(),
         })
     }
 

vortex-layout/src/scan/tasks.rs

@@ -13,9 +13,11 @@ use futures::future::BoxFuture;
 use vortex_array::ArrayRef;
 use vortex_array::MaskFuture;
 use vortex_array::expr::BoundExpression;
+use vortex_array::optimizer::ArrayOptimizer;
 use vortex_error::VortexResult;
 use vortex_mask::Mask;
 use vortex_scan::row_mask::RowMask;
+use vortex_session::VortexSession;
 
 use crate::LayoutReaderRef;
 use crate::scan::filter::FilterExpr;
@@ -50,11 +52,12 @@ pub(crate) fn split_exec(ctx: &TaskContext, read_mask: RowMask) -> VortexResult<
     let projection =
         ctx.reader
             .projection_evaluation(&row_range, &ctx.projection, filter_mask.clone())?;
+    let session = ctx.session.clone();
     Ok(async move {
         if filter_mask.await?.all_false() {
             return Ok(None);
         }
-        projection.await.map(Some)
+        projection.await?.optimize(&session).map(Some)
     }
     .boxed())
 }
@@ -71,7 +74,10 @@ pub(crate) fn split_projection(
     let projection =
         ctx.reader
             .projection_evaluation(row_range, &ctx.projection, MaskFuture::ready(mask))?;
-    Ok(projection.map(|array| array.map(Some)).boxed())
+    let session = ctx.session.clone();
+    Ok(projection
+        .map(move |array| array?.optimize(&session).map(Some))
+        .boxed())
 }
 
 /// Build the filtered mask for a split.
@@ -168,6 +174,10 @@ pub(crate) struct TaskContext {
     pub(crate) reader: LayoutReaderRef,
     /// The projection expression to apply to gather the scanned rows.
     pub(crate) projection: BoundExpression,
+    /// The session the projected arrays are optimized with before the scan returns them. Readers
+    /// apply the projection lazily, so without this a caller would see expression nodes, such as
+    /// a pack, instead of the arrays they reduce to.
+    pub(crate) session: VortexSession,
 }
 
 #[cfg(test)]