[Data] Fix null-type handling in Arrow schema reconciliation and struct alignment

[dragongu]

Description

Fix null-type handling in Arrow schema reconciliation and struct alignment.

A nested field that is all-null in one block and a struct in another crashes _backfill_missing_fields with TypeError: 'pyarrow.lib.DataType' object is not iterable. This PR fixes the nested case and one pre-existing bug.

Bug A — Nested all-null field crashes alignment (the original fix)

_align_struct_fields guards top-level columns with isinstance(column.type, pa.StructType), but its recursive call only inspects the target field type, so a nested pa.null() column slips through.

t1 = pa.table({"outer": pa.array([{"inner": {"y": 1}}])})
t2 = pa.table({"outer": pa.array([{"inner": None}, {"inner": None}])})
concat([t1, t2])
# before: TypeError: 'pyarrow.lib.DataType' object is not iterable
# after:  [{"inner": {"y": 1}}, {"inner": None}, {"inner": None}]

Bug B — _reconcile_field step 4 returns pa.null() instead of list type (pre-existing)

The non_null_types parameter actually received pa.null() entries from the caller. Step 3 (struct reconciliation) had its own per-step null filter, but step 4 (null-list reconciliation) iterated the unfiltered list. When the input was [pa.null(), pa.list_(pa.null()), pa.list_(pa.int64())], step 4 returned pa.null() instead of pa.list_(pa.int64()).

unify_schemas([
    pa.schema([("col", pa.null())]),
    pa.schema([("col", pa.list_(pa.null()))]),
    pa.schema([("col", pa.list_(pa.int64()))]),
])
# before: reconciled to null (wrong)
# after:  reconciled to list<int64>

Fixed by filtering pa.null() once at the top of _reconcile_field instead of per-step, and renaming the parameter from non_null_types to field_types to reflect its actual contract.

Reproduction script

Save as repro.py and run on main before this PR to see both bugs. After applying this PR both print OK.

"""Reproduce two null-handling bugs in Ray Data struct alignment.

Prerequisites:  pip install "ray[data]" pyarrow
Usage:          python repro.py
"""
import ray
import pyarrow as pa
ray.init(runtime_env={"env_vars": {"RAY_MINIMAL": "1"}})
from ray.data._internal.arrow_ops.transform_pyarrow import (
    _align_struct_fields, concat, unify_schemas,
)

# Bug A: nested all-null field crashes with TypeError
print("=== Bug A: nested all-null field ===")
t1 = pa.table({"outer": pa.array([{"inner": {"y": 1}}])})
t2 = pa.table({"outer": pa.array([{"inner": None}, {"inner": None}])})
try:
    result = concat([t1, t2])
    print(f"OK: {result['outer'].to_pylist()}")
except TypeError as e:
    print(f"CRASH: TypeError: {e}")
# Expected: OK: [{'inner': {'y': 1}}, {'inner': None}, {'inner': None}]

# Bug B: null arm shadows concrete list type in _reconcile_field step 4
print("\n=== Bug B: null + list<null> + list<int64> ===")
s1 = pa.schema([("col", pa.null())])
s2 = pa.schema([("col", pa.list_(pa.null()))])
s3 = pa.schema([("col", pa.list_(pa.int64()))])
unified = unify_schemas([s1, s2, s3])
col_type = unified.field("col").type
if col_type == pa.list_(pa.int64()):
    print(f"OK: {col_type}")
else:
    print(f"BUG: expected list<int64>, got {col_type}")
# Expected: OK: list<int64>

[gemini-code-assist]

Code Review

This pull request improves PyArrow schema unification and struct field alignment in Ray Data. Key changes include stripping null types first during field reconciliation, ensuring struct fields are only reconciled when all arms are structs (preventing incorrect unification of struct and primitive mixes), and properly handling all-null nested fields by filling them with nulls of the target struct type. Comprehensive unit tests are also added. The reviewer suggested a performance optimization in _reconcile_field to immediately return the single type when only one non-null type is present, avoiding unnecessary overhead.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit e888075. Configure here.}

[iamjustinhsu]

Wow nice catch! tyty

[richardliaw]

@iamjustinhsu don't forget to click go/enable auto-merge if ready to merge