[Data] Raise a clear error when _backfill_missing_fields gets a non-struct column

[jackylee-ch]

Description

_align_struct_fields guards top-level columns with isinstance(column.type, pa.StructType), but its recursive call only checks the target field type. So a nested field that is a struct in one block and a primitive in another passed the outer guard and failed inside _backfill_missing_fields while iterating column.type. unify_schemas permits the mismatch: _reconcile_field keeps only the struct arms, so the unified type claims inner is a struct even though one block holds an int64.

a = ray.data.from_items([{"c": {"inner": {"y": 1}}}])
b = ray.data.from_items([{"c": {"inner": 5}}])
a.union(b).repartition(1).take_all()
# was: TypeError: 'pyarrow.lib.DataType' object is not iterable
# now: ValueError: Column of type int64 cannot be aligned with struct type
#      struct<y: int64>. A block holds a non-struct value where the unified
#      schema expects a struct.

This adds that same check at the top of the function. An all-null nested field infers as pa.null() and is filled with nulls of the target type rather than rejected.

Raising seemed safer than the issue's other option (backfilling a null struct), since coercing a schema conflict silently alters data — happy to switch.

The issue's repro no longer reaches this path: for a top-level mismatch the outer guard skips alignment and pa.concat_tables raises a readable ArrowTypeError. Only the nested case leaked the TypeError.

Related issues

Fixes #61656

Additional information

Tests: the nested mismatch via _align_struct_fields and via concat, plus the all-null case. unit/test_transform_pyarrow.py 81 passed (was 78); unit/test_arrow_block.py 66, test_map.py 240, test_union.py 5, datasource/test_json.py 58 passed; pre-commit clean; reverting either branch fails its test. Bazel/C++ not run locally.

#61656 is unassigned and no open PR touches this file; two earlier attempts (#62170, #63960) were closed by the stale bot without maintainer objection. AI assistance (Claude) was used; I reviewed every changed line and ran the tests above.

[gemini-code-assist]

Code Review

This pull request adds a check in _backfill_missing_fields to raise a ValueError when a nested non-struct field is aligned with a struct type, preventing unexpected failures during iteration. It also includes unit tests for this scenario. The review feedback points out a critical edge case where a nested field is entirely null (type pa.null()); the new check would incorrectly raise a ValueError for this valid promotion. It is recommended to explicitly handle null types by returning a null array of the target struct type.

[gemini-code-assist]

If a nested field is entirely null in a block, its inferred type in that block's schema might be pa.null(). During schema unification, this null type is successfully promoted to the target StructType from other blocks. However, when aligning the block, _backfill_missing_fields will be recursively called with column having type pa.null().

With the current check, this will raise a ValueError because pa.null() is not a struct type, even though promoting a null nested field to a struct is a perfectly valid operation.

To support this, we should explicitly handle the pa.types.is_null(column.type) case by returning a null array of the target struct type.

    if pa.types.is_null(column.type):
        return pa.nulls(len(column), type=unified_struct_type)

    if not pa.types.is_struct(column.type):
        raise ValueError(
            f"Column of type {column.type} cannot be aligned with struct type "
            f"{unified_struct_type}. A block holds a non-struct value where the "
            "unified schema expects a struct."
        )

[jackylee-ch]

Confirmed — fixed in 1691281. An all-null nested field does infer as pa.null() and gets promoted to the struct type, so the guard was rejecting a valid case. (That path already raised the same TypeError before this PR, so it was broken rather than working — but it should be filled, not rejected.)

Added the pa.types.is_null branch and test_concat_nested_all_null_field; it fails without the branch. unit/test_transform_pyarrow.py 81 passed, test_map.py 240 passed.