[Data] ArrowInvalid error when you backfill missing fields from map tasks (ray-project#60643)

## Description
Try type casting if struct field types mismatch when backfilling missing
fields

## Related issues
Closes ray-project#60628

## Additional information
> Optional: Add implementation details, API changes, usage examples,
screenshots, etc.

---------

Signed-off-by: machichima <nary12321@gmail.com>
Co-authored-by: Balaji Veeramani <balaji@anyscale.com>

Author: machichima

python/ray/data/_internal/arrow_ops/transform_pyarrow.py

@@ -481,10 +481,19 @@ def _backfill_missing_fields(
                     ndim=field_type.ndim
                 )
 
-            # The schema should already be unified by unify_schemas, so types
-            # should be compatible. If not, let the error propagate up.
-            # No explicit casting needed - PyArrow will handle type compatibility
-            # during struct creation or raise appropriate errors.
+            # Handle type mismatches for primitive types
+            # The schema should already be unified by unify_schemas, but
+            # struct field types may still diverge (e.g., int64 vs float64).
+            # Cast the existing array to match the unified struct field type.
+            elif current_array.type != field_type:
+                try:
+                    current_array = current_array.cast(field_type)
+                except pa.ArrowInvalid as e:
+                    raise ValueError(
+                        f"Cannot cast struct field '{field_name}' from "
+                        f"{current_array.type} to {field_type}: {e}"
+                    ) from e
+
             aligned_fields.append(current_array)
         else:
             # If the field is missing, fill with nulls

python/ray/data/tests/test_map_batches.py

@@ -12,6 +12,10 @@
 import pytest
 
 import ray
+from ray.data._internal.arrow_ops.transform_pyarrow import (
+    MIN_PYARROW_VERSION_TYPE_PROMOTION,
+)
+from ray.data._internal.utils.arrow_utils import get_pyarrow_version
 from ray.data.context import DataContext
 from ray.data.dataset import Dataset
 from ray.data.exceptions import UserCodeException
@@ -798,6 +802,42 @@ async def __call__(self, batch):
     assert len(output) == len(expected_output), (len(output), len(expected_output))
 
 
+@pytest.mark.skipif(
+    get_pyarrow_version() < MIN_PYARROW_VERSION_TYPE_PROMOTION,
+    reason="Requires PyArrow >= 14.0.0 for type promotion in nested struct fields",
+)
+def test_map_batches_struct_field_type_divergence(shutdown_only):
+    """Test map_batches with struct fields that have diverging primitive types."""
+
+    def generator_fn(batch):
+        for i, row_id in enumerate(batch["id"]):
+            if i % 2 == 0:
+                # Yield struct with fields (a: int64, b: string)
+                yield {"data": [{"a": 1, "b": "hello"}]}
+            else:
+                # Yield struct with fields (a: float64, c: int32)
+                # Field 'a' has different type, field 'b' missing, field 'c' new
+                yield {"data": [{"a": 1.5, "c": 100}]}
+
+    ds = ray.data.range(4, override_num_blocks=1)
+    ds = ds.map_batches(generator_fn, batch_size=4)
+    result = ds.materialize()
+
+    rows = result.take_all()
+    assert len(rows) == 4
+
+    # Sort to make the order deterministic.
+    rows.sort(key=lambda r: (r["data"]["a"], str(r["data"]["b"])))
+
+    # Rows with a=1.0 (originally int) should have int cast to float, with c=None
+    assert rows[0]["data"] == {"a": 1.0, "b": "hello", "c": None}
+    assert rows[1]["data"] == {"a": 1.0, "b": "hello", "c": None}
+
+    # Rows with a=1.5 should have float a, with b=None
+    assert rows[2]["data"] == {"a": 1.5, "b": None, "c": 100}
+    assert rows[3]["data"] == {"a": 1.5, "b": None, "c": 100}
+
+
 if __name__ == "__main__":
     import sys
 

python/ray/data/tests/test_transform_pyarrow.py

@@ -473,6 +473,64 @@ def test_struct_with_arrow_variable_shaped_tensor_type(
             )
 
 
+@pytest.mark.skipif(
+    get_pyarrow_version() < MIN_PYARROW_VERSION_TYPE_PROMOTION,
+    reason="Requires PyArrow >= 14.0.0 for type promotion in nested struct fields",
+)
+def test_struct_with_diverging_primitive_types():
+    """Test concatenating tables with struct fields that have diverging primitive types.
+
+    This tests the scenario where struct fields have the same name but different
+    primitive types (e.g., int64 vs float64), which requires type promotion.
+    """
+    import pyarrow as pa
+
+    # Table 1: struct with (a: int64, b: string)
+    t1 = pa.table(
+        {
+            "data": pa.array(
+                [{"a": 1, "b": "hello"}, {"a": 2, "b": "world"}],
+                type=pa.struct([pa.field("a", pa.int64()), pa.field("b", pa.string())]),
+            )
+        }
+    )
+
+    # Table 2: struct with (a: float64, c: int32)
+    # Field 'a' has different type, field 'b' missing, field 'c' new
+    t2 = pa.table(
+        {
+            "data": pa.array(
+                [{"a": 1.5, "c": 100}, {"a": 2.5, "c": 200}],
+                type=pa.struct(
+                    [pa.field("a", pa.float64()), pa.field("c", pa.int32())]
+                ),
+            )
+        }
+    )
+
+    # Concatenate with type promotion
+    result = concat([t1, t2], promote_types=True)
+
+    # Verify schema: field 'a' should be promoted to float64
+    expected_struct_type = pa.struct(
+        [
+            pa.field("a", pa.float64()),
+            pa.field("b", pa.string()),
+            pa.field("c", pa.int32()),
+        ]
+    )
+    assert result.schema == pa.schema([pa.field("data", expected_struct_type)])
+
+    # Verify data: int64 values should be cast to float64, missing fields filled with None
+    expected_data = [
+        {"a": 1.0, "b": "hello", "c": None},
+        {"a": 2.0, "b": "world", "c": None},
+        {"a": 1.5, "b": None, "c": 100},
+        {"a": 2.5, "b": None, "c": 200},
+    ]
+    assert result.column("data").to_pylist() == expected_data
+
+
 def test_arrow_concat_object_with_tensor_fails(object_with_tensor_fails_blocks):
     with pytest.raises(ArrowConversionError) as exc_info:
         concat(object_with_tensor_fails_blocks)