[Data] Fix Arrow-backed pandas regressions in preprocessors and pandas batches

[AarryaSaraf]

Description

Ray 2.56 (#63017) made BlockAccessor.to_pandas return Arrow-backed pandas columns
(pd.ArrowDtype) instead of copying into NumPy arrays. That removed a copy, but it also
changed two things downstream code relied on:

1. A missing value became pd.NA instead of np.nan. np.nan answers every
   comparison with False; pd.NA answers "unknown", producing a three-valued mask, and
   it cannot be stored in a numeric NumPy array.
2. Pandas' Arrow backend is not a drop-in for its NumPy backend. %, divmod,
   rmod and rdivmod are unimplemented
   (pandas-dev/pandas#58723), and
   to_numpy()/.values degrade to dtype=object.

Six user-visible regressions followed:

Symptom
PowerTransformer raised IndexError: only integers, slices ... are valid indices — series >= 0 yields a three-valued bool[pyarrow] mask that cannot index a NumPy array.
Categorizer raised TypeError: boolean value of NA is ambiguous — pd.NA survived the null filter into the category sort.
Concatenator silently emitted a column of pickled Python objects instead of a tensor, and raised TypeError: ... not 'NAType' with an integer dtype or flatten=True.
FeatureHasher silently emitted pickled objects too: 0 + pd.NA is pd.NA, so one null poisons every hash bucket.
A pandas UDF doing df["x"] % 2 raised NotImplementedError: mod not implemented.
A pandas UDF doing df[["a", "b"]].to_numpy() silently got dtype=object instead of float64.

Three of the six raised nothing at all, so the new tests assert on Arrow types and dtypes
rather than only on values.

The fix is two independently reviewable commits.

Commit 1 — the four preprocessors. is_null now recognizes pd.NA/pd.NaT, which is
the whole Categorizer fix. PowerTransformer, Concatenator and FeatureHasher
convert the columns they operate on through a new to_numpy_backed helper, which
reproduces the conversion pandas performs when Arrow-backing is disabled and is a no-op on
already-NumPy-backed input.

Commit 2 — the batch-format contract. Ray cannot audit what a user's function does
with a pandas frame, so map_batches and iter_batches now convert pandas batches back to
NumPy-backed pandas before calling it. Ray's own preprocessors are audited (commit 1) and
opt back in through an internal arrow_backed_pandas flag, so they keep the zero-copy
conversion #63017 added. Dataset.to_pandas keeps Arrow-backed dtypes as well: it returns
a result to the driver rather than feeding a user function, and Arrow-backing is what lets
an integer column containing nulls stay integral instead of widening to float.

Caller	pandas backing	Why
Ray's preprocessors	Arrow (zero-copy)	audited for pd.NA, covered by tests
map_batches(batch_format="pandas")	NumPy	arbitrary user code
iter_batches(batch_format="pandas")	NumPy	arbitrary user code
Dataset.to_pandas()	Arrow	driver-side result; preserves types

No public API signature changes: the flag is threaded through the existing
Dataset.map_batches_internal and the private DataIterator._iter_batches.
The pre-2.56 dtype mapping is lossy, and this PR brings that back. An integer column
containing nulls widens to float64, losing exactness above 2**53; pd.NA becomes
np.nan, so a float column no longer distinguishes "missing" from "not a number"; and
bool-with-nulls, string, decimal and nested columns become object. This is deliberate —
it is the representation NumPy-assuming code requires and the one Ray returned in every
release before 2.56. Callers who need exact typesshould use Dataset.to_pandas() or batch_format="pyarrow", neither of which converts.

Related issues

Related to #64765 (the SimpleImputer failure from the same 2.56 change, fixed in #65187 —
this PR adds the regression test that fix shipped without).

Additional information

This is a default-behaviour change. map_batches(batch_format="pandas") and
iter_batches(batch_format="pandas") return NumPy-backed pandas again, as they did before
2.56, so anyone who started depending on Arrow-backed dtypes there in 2.56 will see the
pre-2.56 dtypes instead. That is the intent, but it warrants a look from the owners of
#63017. DataContext.enable_arrow_backed_pandas_conversion remains the escape hatch in the
other direction; its docstring is updated here, because it previously advised setting it to
False specifically to work around the % failure above.

New tests are all in python/ray/data/tests/test_arrow_backed_pandas.py (56 tests): unit
coverage for is_null and to_numpy_backed, per-preprocessor regression tests, and the
contract in the table above. 22 fail without commit 1's source changes and 4 fail without
commit 2's. They build their input with from_items/range rather than from_pandas,
because a pandas block never goes through to_pandas and so never reaches the code these
bugs live in.

[gemini-code-assist]

Code Review

This pull request ensures that pandas batches handed to user-defined functions are NumPy-backed by default, preventing compatibility issues with Arrow-backed pandas (such as missing operators or pd.NA propagation). It introduces a to_numpy_backed utility and an internal arrow_backed_pandas flag to allow Ray's own preprocessors to opt into Arrow-backed pandas for performance. The review feedback highlights a critical performance optimization in to_numpy_backed to avoid reconstructing DataFrames column-by-column when no Arrow-backed columns are present.

[gemini-code-assist]

Calling to_numpy_backed on a pd.DataFrame currently reconstructs the DataFrame column-by-column even if none of the columns are Arrow-backed (pd.ArrowDtype). Since to_numpy_backed is called on every single pandas batch in _format_batch by default, this introduces a significant and unnecessary CPU/memory overhead for standard NumPy-backed pandas datasets.

We can optimize this by checking if any column is actually Arrow-backed first, and returning the DataFrame directly if not.

Suggested change

	if isinstance(data, pd.DataFrame):
	converted = {column: to_numpy_backed(data[column]) for column in data.columns}
	return pd.DataFrame(converted, index=data.index, columns=data.columns)
	if isinstance(data, pd.DataFrame):
	if not any(isinstance(dtype, pd.ArrowDtype) for dtype in data.dtypes):
	return data
	converted = {column: to_numpy_backed(data[column]) for column in data.columns}
	return pd.DataFrame(converted, index=data.index, columns=data.columns)

[cursor]

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

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