vortex-datafusion: Only use schema mapping for casting

When the table schema and the file schema have the same columns but a
different order, then the projection already reorders them and
previously the schema mapper attempted to also fix the ordering, causing
an incorrect result. With this patch, the schema mapper is only used to
perform the expected casting, while leaving the potential reordering to
the projection.

Signed-off-by: Frederic Branczyk <[email protected]>

Author: brancz

vortex-datafusion/src/persistent/opener.rs

@@ -227,8 +227,10 @@ impl FileOpener for VortexOpener {
                 // for schema evolution and divergence between the table's schema and individual files.
                 filter = filter
                     .map(|filter| {
-                        let logical_file_schema =
-                            compute_logical_file_schema(&physical_file_schema, &logical_schema);
+                        let logical_file_schema = compute_logical_file_schema(
+                            &physical_file_schema.clone(),
+                            &logical_schema,
+                        );
 
                         let expr = expr_adapter_factory
                             .create(logical_file_schema, physical_file_schema.clone())
@@ -244,9 +246,14 @@ impl FileOpener for VortexOpener {
                 predicate_file_schema = physical_file_schema.clone();
             }
 
-            let (schema_mapping, adapted_projections) =
+            // Create the initial mapping from physical file schema to projected schema.
+            // This gives us the field reordering and tells us which logical schema fields
+            // to select.
+            let (_schema_mapping, adapted_projections) =
                 schema_adapter.map_schema(&physical_file_schema)?;
 
+            // Build the Vortex projection expression using the adapted projections.
+            // This will reorder the fields to match the target order.
             let fields = adapted_projections
                 .iter()
                 .map(|idx| {
@@ -256,6 +263,39 @@ impl FileOpener for VortexOpener {
                 .collect::<Vec<_>>();
             let projection_expr = select(fields, root());
 
+            // After Vortex applies the projection, the batch will have fields in the target
+            // order (matching adapted_projections), but with the physical file types.
+            // We need a second schema mapping for type casting only, not reordering.
+            // Build a schema that represents what Vortex will return: fields in target order
+            // with physical types.
+            let projected_physical_fields: Vec<Field> = adapted_projections
+                .iter()
+                .map(|&idx| {
+                    let logical_field = logical_schema.field(idx);
+                    let field_name = logical_field.name();
+
+                    // Find this field in the physical schema to get its physical type
+                    physical_file_schema
+                        .field_with_name(field_name)
+                        .map(|phys_field| {
+                            Field::new(
+                                field_name,
+                                merge_field_types(phys_field, logical_field),
+                                phys_field.is_nullable(),
+                            )
+                        })
+                        .unwrap_or_else(|_| (*logical_field).clone())
+                })
+                .collect();
+
+            let projected_physical_schema =
+                Arc::new(arrow_schema::Schema::new(projected_physical_fields));
+
+            // Create a second mapping from the projected physical schema (what Vortex returns)
+            // to the final projected schema. This mapping will handle type casting without reordering.
+            let (batch_schema_mapping, _) =
+                schema_adapter.map_schema(&projected_physical_schema)?;
+
             // We share our layout readers with others partitions in the scan, so we can only need to read each layout in each file once.
             let layout_reader = match layout_reader.entry(file_meta.object_meta.location.clone()) {
                 Entry::Occupied(mut occupied_entry) => {
@@ -350,7 +390,7 @@ impl FileOpener for VortexOpener {
                     ))))
                 })
                 .try_flatten()
-                .map(move |batch| batch.and_then(|b| schema_mapping.map_batch(b)))
+                .map(move |batch| batch.and_then(|b| batch_schema_mapping.map_batch(b)))
                 .boxed();
 
             Ok(stream)