GeezerVison

The World Wide Web, restored to 800 pixels of glory.

GeezerVison is showing github.com — fix: Normalize Arrow-backed dtypes from Ray Dataset.to_pandas() for pyarrow >=25 compatibility by ntkathole · Pull Request #6916 · feast-dev/feast

fix: Normalize Arrow-backed dtypes from Ray Dataset.to_pandas() for pyarrow >=25 compatibility #6916

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

ntkathole merged 3 commits into feast-dev:master from ntkathole:fix_ray_topanda

Oct 1, 2026

Uh oh!

There was an error while loading. Please reload this page.

Merged

fix: Normalize Arrow-backed dtypes from Ray Dataset.to_pandas() for pyarrow >=25 compatibility #6916

Show file tree

Hide file tree

Changes from all commits

Commits

Show all changes

3 commits

Select commit Hold shift + click to select a range

3f7a140

fix: Normalize Arrow-backed dtypes from Ray Dataset.to_pandas() for p…

ntkathole Oct 1, 2026

4dbb478

fix: Detect Ray TensorDtype/PythonObjectDtype by module and normalize…

ntkathole Oct 1, 2026

8850a51

fix: Recover Arrow list value types for empty-list historical features

ntkathole Oct 1, 2026

File filter

Filter by extension .py (5) All 1 file type selected

Viewed files

Clear filters

Conversations

Failed to load comments.

Uh oh!

There was an error while loading. Please reload this page.

Jump to

Jump to file

Failed to load files.

Uh oh!

There was an error while loading. Please reload this page.

Diff view

Unified

Split

Hide whitespace

Diff view

Unified

Split

Hide whitespace

There are no files selected for viewing

15 changes: 11 additions & 4 deletions 15 sdk/python/feast/infra/compute_engines/ray/job.py

Show comments

View file

Open in desktop

Original file line number	Diff line number	Diff line change
Expand Up		@@ -24,6 +24,7 @@
		from feast.infra.offline_stores.file_source import SavedDatasetFileStorage
		from feast.infra.offline_stores.offline_store import RetrievalJob, RetrievalMetadata
		from feast.infra.ray_initializer import get_ray_wrapper
		from feast.infra.ray_shared_utils import normalize_arrow_dtypes
		from feast.repo_config import RepoConfig
		from feast.saved_dataset import SavedDatasetStorage

Expand Down Expand Up		@@ -121,7 +122,9 @@ def to_df(
		assert self._result_dataset is not None, (
		"Dataset should not be None after execution"
		)
		self._result_df = self._result_dataset.to_pandas()
		self._result_df = normalize_arrow_dtypes(
		self._result_dataset.to_pandas()
		)

		# Handle validation if provided
		if validation_reference:
Expand Down Expand Up		@@ -157,7 +160,9 @@ def to_arrow(
		assert self._result_dataset is not None, (
		"Dataset should not be None after execution"
		)
		self._result_arrow = self._result_dataset.to_pandas().to_arrow()
		self._result_arrow = pa.Table.from_pandas(
		normalize_arrow_dtypes(self._result_dataset.to_pandas())
		)

		# Handle validation if provided
		if validation_reference:
Expand Down Expand Up		@@ -249,15 +254,17 @@ def _to_df_internal(self, timeout: Optional[int] = None) -> pd.DataFrame:
		assert self._result_dataset is not None, (
		"Dataset should not be None after execution"
		)
		return self._result_dataset.to_pandas()
		return normalize_arrow_dtypes(self._result_dataset.to_pandas())

		def _to_arrow_internal(self, timeout: Optional[int] = None) -> pa.Table:
		"""Internal method to get Arrow Table (used by parent class)."""
		self._ensure_executed()
		assert self._result_dataset is not None, (
		"Dataset should not be None after execution"
		)
		return self._result_dataset.to_pandas().to_arrow()
		return pa.Table.from_pandas(
		normalize_arrow_dtypes(self._result_dataset.to_pandas())
		)


		@dataclass
Expand Down

73 changes: 37 additions & 36 deletions 73 sdk/python/feast/infra/offline_stores/contrib/ray_offline_store/ray.py

Show comments

View file

Open in desktop

Original file line number	Diff line number	Diff line change
Expand Up		@@ -47,7 +47,9 @@
		apply_field_mapping,
		ensure_timestamp_compatibility,
		is_ray_data,
		normalize_arrow_dtypes,
		normalize_timestamp_columns,
		pandas_to_arrow_with_list_types,
		)
		from feast.infra.registry.base_registry import BaseRegistry
		from feast.on_demand_feature_view import OnDemandFeatureView
Expand All		@@ -59,6 +61,7 @@
		feast_value_type_to_pandas_type,
		pa_to_feast_value_type,
		)
		from feast.types import from_feast_to_pyarrow_type
		from feast.utils import (
		_get_column_names,
		compute_non_entity_date_range,
Expand Down Expand Up		@@ -988,6 +991,7 @@ def __init__(
		self._feature_refs: List[str] = []
		self._entity_df: Optional[pd.DataFrame] = None
		self._prefer_ray_datasets: bool = True
		self._feature_pa_types: Dict[str, pa.DataType] = {}

		def _create_metadata(self) -> RetrievalMetadata:
		"""Create metadata from the entity DataFrame and feature references."""
Expand Down Expand Up		@@ -1084,7 +1088,7 @@ def to_df(
		else:
		if self._prefer_ray_datasets:
		ray_ds = self._get_ray_dataset()
		df = ray_ds.to_pandas()
		df = normalize_arrow_dtypes(ray_ds.to_pandas())
		else:
		result = self._resolve()
		if isinstance(result, pd.DataFrame):
Expand Down Expand Up		@@ -1117,24 +1121,14 @@ def to_arrow(
		validation_reference=validation_reference, timeout=timeout
		)

		if self._prefer_ray_datasets:
		try:
		import ray as _ray

		ray_ds = self._get_ray_dataset()
		return pa.concat_tables(_ray.get(ray_ds.to_arrow_refs()))
		except Exception:
		df = self.to_df(
		validation_reference=validation_reference, timeout=timeout
		)
		return pa.Table.from_pandas(df)
		else:
		result = self._resolve()
		if isinstance(result, pd.DataFrame):
		return pa.Table.from_pandas(result)
		else:
		df = result.to_pandas()
		return pa.Table.from_pandas(df)
		# Always go through the normalized pandas path. Ray's
		# Dataset.to_arrow_refs() preserves TensorDtype / PythonObjectDtype as
		# ArrowPythonObjectType(large_binary), which breaks list-feature type
		# contracts. normalize_arrow_dtypes() in to_df() restores plain Python
		# lists / ndarrays; feature Arrow types (when set) recover list value
		# types for all-empty list columns (list<item: null> → list<item: T>).
		df = self.to_df(validation_reference=validation_reference, timeout=timeout)
		return pandas_to_arrow_with_list_types(df, self._feature_pa_types)

		def to_feast_df(
		self,
Expand Down Expand Up		@@ -1191,27 +1185,14 @@ def to_sql(self) -> str:
		def _to_df_internal(self, timeout: Optional[int] = None) -> pd.DataFrame:
		if self._prefer_ray_datasets:
		ray_ds = self._get_ray_dataset()
		return ray_ds.to_pandas()
		return normalize_arrow_dtypes(ray_ds.to_pandas())
		else:
		return self._resolve().to_pandas()

		def _to_arrow_internal(self, timeout: Optional[int] = None) -> pa.Table:
		if self._prefer_ray_datasets:
		try:
		import ray as _ray

		ray_ds = self._get_ray_dataset()
		return pa.concat_tables(_ray.get(ray_ds.to_arrow_refs()))
		except Exception:
		ray_ds = self._get_ray_dataset()
		return pa.Table.from_pandas(ray_ds.to_pandas())
		else:
		result = self._resolve()
		if isinstance(result, pd.DataFrame):
		return pa.Table.from_pandas(result)
		else:
		df = result.to_pandas()
		return pa.Table.from_pandas(df)
		return pandas_to_arrow_with_list_types(
		self._to_df_internal(timeout=timeout), self._feature_pa_types
		)

		def persist(
		self,
Expand Down Expand Up		@@ -2628,5 +2609,25 @@ def finalize_result(batch: pd.DataFrame) -> pd.DataFrame:
		job._on_demand_feature_views = on_demand_feature_views
		job._feature_refs = feature_refs
		job._entity_df = entity_df_sample
		# Preserve Feast-declared Arrow types so to_arrow() can recover
		# list<item: T> when a result column only contains empty lists.
		# Promote list<float32> → list<float64>: the pandas materialization
		# path historically returns float64 for float features, and the
		# universal type tests assert pa.types.is_float64.
		feature_pa_types: Dict[str, pa.DataType] = {}
		for fv in regular_feature_views:
		for feature in fv.features:
		try:
		pa_type = from_feast_to_pyarrow_type(feature.dtype)
		except (ValueError, KeyError, TypeError):
		continue
		if pa.types.is_list(pa_type) and pa.types.is_float32(
		pa_type.value_type
		):
		pa_type = pa.list_(pa.float64())
		feature_pa_types[feature.name] = pa_type
		if full_feature_names:
		feature_pa_types[f"{fv.name}__{feature.name}"] = pa_type
		job._feature_pa_types = feature_pa_types
		job._metadata = job._create_metadata()
		return job

146 changes: 146 additions & 0 deletions 146 sdk/python/feast/infra/ray_shared_utils.py

Show comments

View file

Open in desktop

Original file line number	Diff line number	Diff line change
Expand Up		@@ -7,6 +7,152 @@
		from ray.data import Dataset


		def _is_ray_extension_dtype(dtype: Any) -> bool:
		"""Return True for Ray-specific pandas extension dtypes.

		Ray Data registers custom extension types such as ``TensorDtype`` and
		``PythonObjectDtype``. These must not be identified via
		``pd.api.types.is_numeric_dtype`` / ``is_object_dtype`` because
		``TensorDtype(dtype=int64)`` reports as numeric and would otherwise be
		skipped.
		"""
		module = type(dtype).__module__
		return module.startswith("ray.") or ".ray." in module


		def _is_nested_arrow_dtype(dtype: Any) -> bool:
		"""Return True for ArrowDtype columns that hold list/struct/map values."""
		arrow_dtype_cls = getattr(pd, "ArrowDtype", None)
		if arrow_dtype_cls is None or not isinstance(dtype, arrow_dtype_cls):
		return False
		pa_type = dtype.pyarrow_dtype
		return (
		pa.types.is_list(pa_type)
		or pa.types.is_large_list(pa_type)
		or pa.types.is_fixed_size_list(pa_type)
		or pa.types.is_struct(pa_type)
		or pa.types.is_map(pa_type)
		)


		def _cell_to_python(value: Any) -> Any:
		"""Convert a Ray extension cell to a plain Python / NumPy object.

		Ray's ``TensorDtype`` yields ``TensorArrayElement`` values that print like
		``array([...])`` but are not ``isinstance(..., np.ndarray)``.

		Empty list features are stored by Feast as typed empty NumPy arrays
		(see ``convert_array_column``) so that Arrow can infer ``list<item: T>``
		rather than ``list<item: null>``. Therefore we prefer keeping values as
		``np.ndarray`` (including empty ones) over calling ``.tolist()``, which
		would turn ``np.empty(0, dtype=int32)`` into a typeless ``[]``.
		"""
		if value is None or isinstance(value, (str, bytes, dict)):
		return value
		if isinstance(value, list):
		return value
		if isinstance(value, tuple):
		return list(value)
		if isinstance(value, np.ndarray):
		return value
		# Ray TensorArrayElement and similar array-likes
		if hasattr(value, "__array__") and not isinstance(value, (pd.Timestamp,)):
		try:
		arr = np.asarray(value)
		if getattr(arr, "ndim", 0) >= 1:
		return arr
		except Exception:
		pass
		if hasattr(value, "tolist") and not isinstance(value, (pd.Timestamp,)):
		try:
		return value.tolist()
		except Exception:
		pass
		return value


		def pandas_to_arrow_with_list_types(
		df: pd.DataFrame,
		column_pa_types: Optional[Dict[str, pa.DataType]] = None,
		) -> pa.Table:
		"""Convert a DataFrame to Arrow, preserving list value types when possible.

		``pa.Table.from_pandas`` infers ``list<item: null>`` when a list column
		contains only empty Python lists. When ``column_pa_types`` provides the
		Feast-declared Arrow type for a column (e.g. ``list<int32>``), that type
		is used to cast the column after conversion.

		List columns inferred as ``list<float32>`` are promoted to ``list<float64>``
		to match the historical pandas materialization path that universal type
		tests assert against.
		"""
		table = pa.Table.from_pandas(df, preserve_index=False)
		column_pa_types = column_pa_types or {}

		arrays = []
		fields = []
		for field in table.schema:
		col = table.column(field.name)
		target = column_pa_types.get(field.name)
		col_type = field.type

		if (
		target is not None
		and pa.types.is_list(target)
		and pa.types.is_list(col_type)
		and pa.types.is_null(col_type.value_type)
		and not pa.types.is_null(target.value_type)
		):
		col = col.cast(target)
		col_type = target
		elif pa.types.is_list(col_type) and pa.types.is_float32(col_type.value_type):
		col_type = pa.list_(pa.float64())
		col = col.cast(col_type)

		arrays.append(col)
		fields.append(pa.field(field.name, col_type, nullable=field.nullable))
		return pa.Table.from_arrays(arrays, schema=pa.schema(fields))


		def normalize_arrow_dtypes(df: pd.DataFrame) -> pd.DataFrame:
		"""Convert Ray/Arrow nested extension dtypes to traditional object columns.

		Ray Data >= 2.58 with pyarrow >= 25 may produce extension dtype columns
		from ``Dataset.to_pandas()``:

		* ``pd.ArrowDtype(list<...>)`` — Arrow-backed list types
		* ``PythonObjectDtype`` — Ray's generic object extension for ragged lists
		* ``TensorDtype`` — Ray's fixed-shape tensor extension for uniform lists

		Downstream Feast code and universal type tests expect list feature columns
		as plain ``object`` dtype containing Python ``list`` / ``np.ndarray``
		values, and Arrow tables with real ``list`` types (not
		``ArrowPythonObjectType(large_binary)``).

		Scalar pandas / Arrow extension dtypes (``Int64Dtype``,
		``ArrowDtype(int64)``, etc.) are left untouched.
		"""
		cols_to_fix = [
		c
		for c in df.columns
		if isinstance(df[c].dtype, pd.api.extensions.ExtensionDtype)
		and (
		_is_ray_extension_dtype(df[c].dtype) or _is_nested_arrow_dtype(df[c].dtype)
		)
		]
		if not cols_to_fix:
		return df

		df = df.copy()
		for col in cols_to_fix:
		df[col] = pd.Series(
		[_cell_to_python(v) for v in df[col]],
		index=df.index,
		dtype=object,
		)
		return df


		class RemoteDatasetProxy:
		"""Proxy class that executes Ray Data operations remotely on cluster workers."""

Expand Down

Oops, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

Oops, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

Add this suggestion to a batch that can be applied as a single commit. This suggestion is invalid because no changes were made to the code. Suggestions cannot be applied while the pull request is closed. Suggestions cannot be applied while viewing a subset of changes. Only one suggestion per line can be applied in a batch. Add this suggestion to a batch that can be applied as a single commit. Applying suggestions on deleted lines is not supported. You must change the existing code in this line in order to create a valid suggestion. Outdated suggestions cannot be applied. This suggestion has been applied or marked resolved. Suggestions cannot be applied from pending reviews. Suggestions cannot be applied on multi-line comments. Suggestions cannot be applied while the pull request is queued to merge. Suggestion cannot be applied right now. Please check back later.

You are visitor 01402503 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.