Add resolve_batches, check_unify_batch_dtype, and entry-type From impls

src/enums/array.rs

@@ -2810,6 +2810,103 @@ impl Array {
         }
     }
 
+    /// Checks whether this array's dtype matches the target Field, and
+    /// converts it when possible using the existing cast methods.
+    ///
+    /// The function exists to support the `allow_mixed_array_batches`
+    /// mode which relaxes uniform `SuperArray` typing restrictions.
+    /// When this method is ran, it coerces to the target `Field` type.
+    ///
+    /// ## Conversion rules
+    ///
+    /// | Input | Result |
+    /// |---|---|
+    /// | Matching dtype | Original array, without copying |
+    /// | Different numeric width | Corresponding `NumericArray::try_i32`, `try_i64`, `try_u32`, `try_u64`, `try_f32`, or `try_f64` conversion |
+    /// | Datetime32 to Datetime64 | `TemporalArray::try_dt64` |
+    /// | Datetime64 to Datetime32 | `TemporalArray::try_dt32` |
+    /// | String32 to String64 (`LargeString`) | `From<&StringArray<u32>> for StringArray<u64>` |
+    /// | Unsupported conversion | All-null array with the target dtype and source length, created through `Array::null_array` |
+    /// | Empty input | Zero-row array with the target field dtype |
+    #[cfg(feature = "allow_mixed_array_batches")]
+    pub fn check_unify_batch_dtype(array: Array, field: &crate::Field) -> Array {
+        use crate::ffi::arrow_dtype::ArrowType;
+
+        let len = array.len();
+
+        // Empty array: zero-row array of the Field dtype.
+        if len == 0 {
+            return Array::from_arrow_dtype(&field.dtype);
+        }
+
+        // Dtype already matches: pass through with no copy.
+        if array.arrow_type() == field.dtype {
+            return array;
+        }
+
+        let converted = match (&array, &field.dtype) {
+            // Numeric array at another numeric width.
+            (Array::NumericArray(num), ArrowType::Int32) => {
+                num.try_i32().ok().map(|a| Array::NumericArray(NumericArray::Int32(a)))
+            }
+            (Array::NumericArray(num), ArrowType::Int64) => {
+                num.try_i64().ok().map(|a| Array::NumericArray(NumericArray::Int64(a)))
+            }
+            (Array::NumericArray(num), ArrowType::UInt32) => {
+                num.try_u32().ok().map(|a| Array::NumericArray(NumericArray::UInt32(a)))
+            }
+            (Array::NumericArray(num), ArrowType::UInt64) => {
+                num.try_u64().ok().map(|a| Array::NumericArray(NumericArray::UInt64(a)))
+            }
+            (Array::NumericArray(num), ArrowType::Float32) => {
+                num.try_f32().ok().map(|a| Array::NumericArray(NumericArray::Float32(a)))
+            }
+            (Array::NumericArray(num), ArrowType::Float64) => {
+                num.try_f64().ok().map(|a| Array::NumericArray(NumericArray::Float64(a)))
+            }
+
+            // Datetime32 under a Datetime64 Field.
+            #[cfg(feature = "datetime")]
+            (Array::TemporalArray(temp), target)
+                if matches!(
+                    target,
+                    ArrowType::Date64
+                        | ArrowType::Time64(_)
+                        | ArrowType::Duration64(_)
+                        | ArrowType::Timestamp(_, _)
+                ) =>
+            {
+                temp.try_dt64()
+                    .ok()
+                    .map(|a| Array::TemporalArray(crate::TemporalArray::Datetime64(a)))
+            }
+
+            // Datetime64 under a Datetime32 Field.
+            #[cfg(feature = "datetime")]
+            (Array::TemporalArray(temp), target)
+                if matches!(
+                    target,
+                    ArrowType::Date32 | ArrowType::Time32(_) | ArrowType::Duration32(_)
+                ) =>
+            {
+                temp.try_dt32()
+                    .ok()
+                    .map(|a| Array::TemporalArray(crate::TemporalArray::Datetime32(a)))
+            }
+
+            // String32 under a String64 (LargeString) Field.
+            #[cfg(feature = "large_string")]
+            (Array::TextArray(crate::TextArray::String32(s32)), ArrowType::LargeString) => {
+                let widened = crate::StringArray::<u64>::from(&**s32);
+                Some(Array::TextArray(crate::TextArray::String64(Arc::new(widened))))
+            }
+
+            _ => None,
+        };
+
+        converted.unwrap_or_else(|| Array::null_array(&field.dtype, len))
+    }
+
     /// Build an array from a slice of Scalars.
     ///
     /// All scalars must be the same type. The type is inferred from the first

src/structs/chunked/super_array.rs

@@ -45,11 +45,11 @@
 //! Field-based constructors and `push_field_array` also provide `try_*`
 //! variants that return `Result` when runtime data may not match the required
 //! type.
-//! 
+//!
 //! Mixed cases cannot be used in `SuperTable` and are rejected at the boundary
 //! as it would violate contractual `Field`-based guarantees. Hence, these are
-//! intended for transient workloads only. 
-//! 
+//! intended for transient workloads only.
+//!
 //! ## Apache Arrow / Polars bridges (`cast_arrow` / `cast_polars` features)
 //! - `to_apache_arrow()` exports each chunk as an arrow-rs `ArrayRef`.
 //! - `to_polars()` builds a polars `Series` whose internal chunks mirror the SuperArray.
@@ -258,7 +258,7 @@ impl SuperArray {
     /// Constructs a SuperArray from raw `Array` chunks with null counts.
     ///
     /// # Panics
-    /// 1. If null_counts length does not match chunks length. 
+    /// 1. If null_counts length does not match chunks length.
     /// 2. On mismatched chunk types, unless the `allow_mixed_array_batches`feature is on.
     pub fn from_arrays_nc(chunks: Vec<Array>, null_counts: Vec<usize>) -> Self {
         assert_eq!(
@@ -1139,6 +1139,70 @@ impl SuperArray {
         chunks.all(|chunk| chunk.arrow_type() == dtype)
     }
 
+    /// Resolves every chunk to the container's target dtype.
+    ///
+    /// ## Behaviour
+    /// - The target dtype is the Field when present, otherwise the type of
+    ///   the first batch.
+    /// - The result ends up with a Field that describes every chunk.
+    /// - For a field-free SuperArray, a Field is constructed from the first chunk's type.
+    /// - Chunks that already match the target pass through without penalty.
+    /// - The conversion per chunk delegates to `Array::check_unify_batch_dtype`.
+    /// - Categorical dictionaries are merged across the resolved chunks
+    ///   via `rebuild_category_manager` when the `shared_dict` feature is on.
+    #[cfg(all(feature = "allow_mixed_array_batches", feature = "views"))]
+    pub fn resolve_batches(self) -> SuperArray {
+        if self.chunks.is_empty() {
+            return self;
+        }
+
+        // Target dtype is the Field when present, otherwise the first chunk.
+        let field = match self.field.clone() {
+            Some(f) => f,
+            None => {
+                let first = &self.chunks[0];
+                Arc::new(Field::new(
+                    "data",
+                    first.arrow_type(),
+                    first.is_nullable(),
+                    None,
+                ))
+            }
+        };
+
+        // Fast path - all chunks already match the target.
+        if self
+            .chunks
+            .iter()
+            .all(|chunk| chunk.arrow_type() == field.dtype)
+        {
+            let mut sa = self;
+            if sa.field.is_none() {
+                sa.field = Some(field);
+            }
+            return sa;
+        }
+
+        // Convert each chunk through check_unify_batch_dtype.
+        let chunks: Vec<Array> = self
+            .chunks
+            .into_iter()
+            .map(|chunk| Array::check_unify_batch_dtype(chunk, &field))
+            .collect();
+
+        #[cfg_attr(not(feature = "shared_dict"), allow(unused_mut))]
+        let mut sa = SuperArray {
+            chunks,
+            field: Some(field),
+            null_counts: None,
+            #[cfg(feature = "shared_dict")]
+            category_manager: None,
+        };
+        #[cfg(feature = "shared_dict")]
+        sa.rebuild_category_manager();
+        sa
+    }
+
     /// Borrow the column's `CategoryManagerT`, or `None` if the column
     /// is not categorical or no chunks have been pushed yet.
     ///
@@ -1250,6 +1314,19 @@ impl FromIterator<Array> for SuperArray {
     }
 }
 
+/// Consolidates all chunks into a single contiguous `ArrayV`.
+///
+/// Empty SuperArrays produce a zero-row array of the Field dtype
+/// when a Field is present, or `Array::Null` otherwise.
+#[cfg(feature = "views")]
+impl From<SuperArray> for ArrayV {
+    fn from(sa: SuperArray) -> Self {
+        #[cfg(feature = "allow_mixed_array_batches")]
+        let sa = sa.resolve_batches();
+        ArrayV::from(sa.consolidate())
+    }
+}
+
 impl Shape for SuperArray {
     fn shape(&self) -> ShapeDim {
         ShapeDim::Rank1(self.len())
@@ -1731,8 +1808,9 @@ mod tests {
         assert_eq!(sa.len(), 5);
     }
 
-    /// A `Field`-carrying constructor rejects mixed chunks with the feature
-    /// on: a present field always describes every chunk.
+    /// - `Field` holding constructor rejects mixed chunks with the feature
+    /// on. 
+    /// - Presents field instance describes every chunk.
     #[cfg(feature = "allow_mixed_array_batches")]
     #[test]
     #[should_panic(expected = "ArrowType mismatch")]
@@ -1746,7 +1824,7 @@ mod tests {
         );
     }
 
-    /// A push onto a `Field`-carrying SuperArray rejects a mismatched chunk
+    /// A push onto a `Field`-holding SuperArray rejects a mismatched chunk
     /// with the feature on.
     #[cfg(feature = "allow_mixed_array_batches")]
     #[test]

src/structs/chunked/super_table.rs

@@ -1199,6 +1199,16 @@ impl From<SuperTableV> for SuperTable {
     }
 }
 
+/// Consolidates all batches into a single `TableV`.
+///
+/// Empty SuperTables produce an empty `TableV` via `Table::default`.
+#[cfg(feature = "views")]
+impl From<SuperTable> for TableV {
+    fn from(st: SuperTable) -> Self {
+        TableV::from(st.consolidate())
+    }
+}
+
 /// Ergonomic constructor for a [`SuperTable`] from named table batches.
 ///
 /// Each batch argument may be a `Table` or `Arc<Table>`; both flow