Revert "refactor: simplify cast_struct_column and extract helpers"

This reverts commit 5a549ec.

Author: kosiew

datafusion/common/src/nested_struct.rs

@@ -55,86 +55,64 @@ fn cast_struct_column(
     target_fields: &[Arc<Field>],
     cast_options: &CastOptions,
 ) -> Result<ArrayRef> {
-    // Fast path for completely-null arrays or a Null-typed column
-    if is_all_null_array(source_col) {
+    if source_col.data_type() == &DataType::Null
+        || (!source_col.is_empty() && source_col.null_count() == source_col.len())
+    {
         return Ok(new_null_array(
             &Struct(target_fields.to_vec().into()),
             source_col.len(),
         ));
     }
 
-    // Ensure the source is actually a StructArray
-    let source_struct = match source_col.as_any().downcast_ref::<StructArray>() {
-        Some(s) => s,
-        None => {
-            return _plan_err!(
-                "Cannot cast column of type {} to struct type: source is not a StructArray.",
-                source_col.data_type()
-            );
-        }
-    };
+    if let Some(source_struct) = source_col.as_any().downcast_ref::<StructArray>() {
+        let source_fields = source_struct.fields();
+        let has_overlap = fields_have_name_overlap(source_fields, target_fields);
+        validate_struct_compatibility(source_fields, target_fields)?;
 
-    let source_fields = source_struct.fields();
-    let has_overlap = fields_have_name_overlap(source_fields, target_fields);
-
-    // Validate that individual fields are cast-compatible before doing work
-    validate_struct_compatibility(source_fields, target_fields)?;
+        let mut fields: Vec<Arc<Field>> = Vec::with_capacity(target_fields.len());
+        let mut arrays: Vec<ArrayRef> = Vec::with_capacity(target_fields.len());
+        let num_rows = source_col.len();
 
-    let num_rows = source_col.len();
+        // Iterate target fields and pick source child either by name (when fields overlap)
+        // or by position (when there is no name overlap).
+        for (index, target_child_field) in target_fields.iter().enumerate() {
+            fields.push(Arc::clone(target_child_field));
 
-    // Collect target field descriptors (we preserve exact target Field objects)
-    let fields: Vec<Arc<Field>> = target_fields.iter().cloned().collect();
+            // Determine the source child column: by name when overlapping names exist,
+            // otherwise by position.
+            let source_child_opt: Option<&ArrayRef> = if has_overlap {
+                source_struct.column_by_name(target_child_field.name())
+            } else {
+                Some(source_struct.column(index))
+            };
 
-    // Build child arrays in a single iterator to make error propagation concise
-    let arrays: Vec<ArrayRef> = target_fields
-        .iter()
-        .enumerate()
-        .map(|(index, target_child_field)| {
-            match select_source_child(
-                source_struct,
-                has_overlap,
-                index,
-                target_child_field,
-            ) {
+            match source_child_opt {
                 Some(source_child_col) => {
-                    cast_column(source_child_col, target_child_field, cast_options)
-                        .map_err(|e| {
+                    let adapted_child =
+                        cast_column(source_child_col, target_child_field, cast_options)
+                            .map_err(|e| {
                             e.context(format!(
                                 "While casting struct field '{}'",
                                 target_child_field.name()
                             ))
-                        })
+                        })?;
+                    arrays.push(adapted_child);
+                }
+                None => {
+                    arrays.push(new_null_array(target_child_field.data_type(), num_rows));
                 }
-                None => Ok(new_null_array(target_child_field.data_type(), num_rows)),
             }
-        })
-        .collect::<Result<Vec<_>>>()?;
-
-    let struct_array =
-        StructArray::new(fields.into(), arrays, source_struct.nulls().cloned());
-    Ok(Arc::new(struct_array))
-}
-
-/// Helper: return true if column is a Null type or contains only nulls
-fn is_all_null_array(source_col: &ArrayRef) -> bool {
-    source_col.data_type() == &DataType::Null
-        || (!source_col.is_empty() && source_col.null_count() == source_col.len())
-}
+        }
 
-/// Helper: select a child column by name when names overlap, otherwise by position.
-///
-/// Safety: when `has_overlap == false` callers must ensure `index` is valid; caller guarantees
-/// this via validation earlier (`validate_struct_compatibility`).
-fn select_source_child<'a>(
-    source_struct: &'a StructArray,
-    has_overlap: bool,
-    index: usize,
-    target_child_field: &Field,
-) -> Option<&'a ArrayRef> {
-    if has_overlap {
-        source_struct.column_by_name(target_child_field.name())
+        let struct_array =
+            StructArray::new(fields.into(), arrays, source_struct.nulls().cloned());
+        Ok(Arc::new(struct_array))
     } else {
-        Some(source_struct.column(index))
+        // Return error if source is not a struct type
+        _plan_err!(
+            "Cannot cast column of type {} to struct type. Source must be a struct to cast to struct.",
+            source_col.data_type()
+        )
     }
 }