[Data] Preserve schema for empty datasets in schema unification and to_pandas()

[lonexreb]

Why are these changes needed?

Dataset.to_pandas() (and schema propagation generally) drops columns for an empty dataset:

import pyarrow as pa, ray
empty_ref = ray.put(pa.table([pa.array([], pa.int32())], ["apples"]))
rd = ray.data.from_arrow_refs([empty_ref])
assert list(rd.to_pandas().columns) == ["apples"]  # fails on master: columns == []

Two layers cause this:

1. Metadata layer: unify_block_metadata_schema() / unify_ref_bundles_schema() skip blocks with num_rows == 0, so a dataset made only of empty blocks reports no schema — even though an empty Arrow table carries a perfectly valid one (as noted by @mango766 on the issue). This PR makes empty-block schemas a fallback when no non-empty block provides a schema; non-empty blocks still take precedence.
2. Consumption layer: to_pandas() builds its result from iter_batches(), which yields no batches for an empty dataset, so PandasBlockBuilder.build() produced a DataFrame with zero columns. This PR reconstructs columns and dtypes from the dataset schema, routing Arrow-backed schemas through the same BlockAccessor.to_pandas() path used for non-empty blocks so empty and non-empty results agree on dtypes.

This supersedes #64721 (same fix for layer 2, closed by the stale bot after both bot-review comments had been addressed; GitHub would not allow reopening it). This PR carries those commits, rebased on current master, plus the new metadata-layer fix.

Related issue number

Fixes #59946

Checks

• I've signed off every commit (DCO).
• I've run pre-commit formatting (black==22.10.0 per repo pin) on changed files.
• Testing strategy:
  • Added test_to_pandas_empty_dataset_preserves_columns (Arrow-backed and pandas-backed empty datasets, dtype parity with non-empty datasets).
  • Added test_unify_block_metadata_schema_all_empty_blocks (fallback + precedence).
  • Ran locally against master with the Ray nightly wheel + setup-dev.py symlinks:
    • pytest python/ray/data/tests/test_util.py → 39 passed
    • pytest python/ray/data/tests/test_arrow_block.py -k to_pandas → 13 passed
    • Issue's exact repro passes.

Notes for reviewers / AI-assistance disclosure

• This PR was prepared with AI assistance; I have reviewed every changed line and run the tests above locally.
• Duplicate-work check: no other open PR references [Data] from_arrow_refs converts an empty table with one column into an empty table with zero columns #59946; the only prior PR was my own [Data] Preserve schema columns in to_pandas() for empty datasets #64721, which this supersedes.

[gemini-code-assist]

Code Review

This pull request addresses issue #59946 by ensuring that empty datasets preserve their schema's columns and dtypes when converted to pandas DataFrames, and that schemas are correctly unified when all blocks or bundles are empty. The review feedback highlights critical issues in the implementation: pyarrow.Schema lacks an empty_table() method, which will cause an AttributeError, and retrieving column types from base_schema for pandas-backed datasets will fail and fall back to losing dtypes. Additionally, the reviewer suggests caching bundle.num_rows() to avoid duplicate calls and adding a test case for a 0-block empty dataset to ensure full coverage of the fallback path.

[gemini-code-assist]

The pyarrow.Schema class does not have an empty_table() method. Calling base_schema.empty_table() will raise an AttributeError when executing this path (e.g., for a 0-block empty dataset like ray.data.range(0)). Instead, use pyarrow.Table.from_batches([], schema=base_schema) to create an empty table with the given schema.

Suggested change

	base_schema = getattr(schema, "base_schema", None)
	if isinstance(base_schema, pa.Schema):
	# Route the empty Arrow table through the same
	# `BlockAccessor.to_pandas()` path used for non-empty blocks
	# so the two agree on column types (types_mapper, tensor
	# casting, etc.).
	df = BlockAccessor.for_block(base_schema.empty_table()).to_pandas()
	base_schema = getattr(schema, "base_schema", None)
	if isinstance(base_schema, pa.Schema):
	# Route the empty Arrow table through the same
	# `BlockAccessor.to_pandas()` path used for non-empty blocks
	# so the two agree on column types (types_mapper, tensor
	# casting, etc.).
	df = BlockAccessor.for_block(pa.Table.from_batches([], schema=base_schema)).to_pandas()

[gemini-code-assist]

For pandas-backed datasets, base_schema is typically a pandas.Index or pandas.Series of dtypes, neither of which has a types attribute. Thus, getattr(base_schema, "types", None) will return None, causing it to always fall back to the else block and lose the dtypes. Instead, use schema.types directly, which is a standard property of the Ray Schema class and is guaranteed to contain the column types.

Suggested change

	# Pandas-backed schema: preserve per-column dtypes when known.
	types = getattr(base_schema, "types", None)
	if types is not None and len(types) == len(schema.names):
	df = pandas.DataFrame(
	{
	name: pandas.Series([], dtype=dtype)
	for name, dtype in zip(schema.names, types)
	}
	)
	else:
	df = pandas.DataFrame(columns=list(schema.names))
	# Pandas-backed schema: preserve per-column dtypes when known.
	types = schema.types
	if types and len(types) == len(schema.names):
	df = pandas.DataFrame(
	{
	name: pandas.Series([], dtype=dtype)
	for name, dtype in zip(schema.names, types)
	}
	)
	else:
	df = pandas.DataFrame(columns=list(schema.names))

[gemini-code-assist]

Calling bundle.num_rows() twice in the condition can be optimized by caching the result in a local variable num_rows.

Suggested change

	for bundle in ref_bundles:
	if bundle.schema is not None and (
	bundle.num_rows() is None or bundle.num_rows() > 0
	):
	if bundle.schema is None:
	continue
	if bundle.num_rows() is None or bundle.num_rows() > 0:
	schemas_to_unify.append(bundle.schema)
	else:
	empty_bundle_schemas.append(bundle.schema)
	for bundle in ref_bundles:
	if bundle.schema is None:
	continue
	num_rows = bundle.num_rows()
	if num_rows is None or num_rows > 0:
	schemas_to_unify.append(bundle.schema)
	else:
	empty_bundle_schemas.append(bundle.schema)

[gemini-code-assist]

The current tests only cover empty datasets with 1 empty block (since from_arrow_refs and from_pandas with empty tables/DataFrames still create 1 block). In these cases, df.shape[1] is not 0, so the fallback code path in to_pandas() is never executed. To ensure the fallback code path is actually tested and works (especially for 0-block datasets), we should add test cases for 0-block datasets (e.g., using ray.data.range(0)).

Suggested change

	# Pandas-backed empty datasets preserve columns and dtypes too.
	pandas_df = pd.DataFrame(
	{"a": pd.Series([], dtype="int64"), "b": pd.Series([], dtype="float64")}
	)
	ds3 = ray.data.from_pandas(pandas_df)
	df3 = ds3.to_pandas()
	assert list(df3.columns) == ["a", "b"]
	assert df3["a"].dtype == np.int64
	assert df3["b"].dtype == np.float64
	# Pandas-backed empty datasets preserve columns and dtypes too.
	pandas_df = pd.DataFrame(
	{"a": pd.Series([], dtype="int64"), "b": pd.Series([], dtype="float64")}
	)
	ds3 = ray.data.from_pandas(pandas_df)
	df3 = ds3.to_pandas()
	assert list(df3.columns) == ["a", "b"]
	assert df3["a"].dtype == np.int64
	assert df3["b"].dtype == np.float64
	# Test 0-block empty dataset (Arrow-backed) to ensure the fallback path is covered.
	ds_zero = ray.data.range(0)
	df_zero = ds_zero.to_pandas()
	assert list(df_zero.columns) == ["id"]
	assert df_zero["id"].dtype == np.int64

[cursor]

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

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

[lonexreb]

Thanks for the reviews — addressed in 682e386. Point-by-point:

• "pyarrow.Schema lacks an empty_table() method" (gemini): pa.Schema.empty_table() is a real PyArrow API (available well before Ray's pyarrow >= 17 floor; verified on 24.0.0/25.0.1). test_to_pandas_empty_dataset_preserves_columns exercises exactly this path and passes, including the dtype-parity assertion against a non-empty dataset. No change made.
• "base_schema lacks a types attribute for pandas-backed datasets" (gemini): for pandas-backed datasets Schema.base_schema is a PandasBlockSchema, which does have .types (the test asserts the preserved int64/float64 dtypes). The access is also getattr-guarded with a column-name fallback, so a schema without types degrades to names-only rather than failing.
• Cache bundle.num_rows() (gemini): done.
• 0-block coverage (gemini): added — ray.data.range(0).to_pandas() returns an empty DataFrame cleanly.
• Pandas-backed empty branch bypasses BlockAccessor conversion (cursor): fair point — the empty pandas DataFrame is now routed through the same BlockAccessor.to_pandas() conversion as non-empty blocks, so internal extension dtypes (e.g. TensorDtype) convert identically, with a guard for dtypes that can't back an empty Series.

All 42 tests in test_util.py + the regression test pass locally against master.