import copy

import functools

import uuid

import warnings

from types import FunctionType

from typing import Any, List, Optional, Union, cast

import dill

import pyarrow

from typeguard import typechecked

from feast.aggregation import Aggregation

from feast.base_feature_view import BaseFeatureView

from feast.data_source import RequestSource

from feast.entity import Entity

from feast.errors import RegistryInferenceFailure, SpecifiedFeaturesNotPresentError

from feast.feature_view import DUMMY_ENTITY_NAME, FeatureView, FeatureViewState

from feast.feature_view_projection import FeatureViewProjection

from feast.field import Field, from_value_type

from feast.proto_utils import transformation_to_proto

from feast.protos.feast.core.OnDemandFeatureView_pb2 import (

OnDemandFeatureView as OnDemandFeatureViewProto,

)

from feast.protos.feast.core.OnDemandFeatureView_pb2 import (

OnDemandFeatureViewMeta,

OnDemandFeatureViewSpec,

OnDemandSource,

)

from feast.protos.feast.core.Transformation_pb2 import (

UserDefinedFunctionV2 as UserDefinedFunctionProto,

)

from feast.transformation.base import Transformation

from feast.transformation.mode import TransformationMode

from feast.transformation.pandas_transformation import PandasTransformation

from feast.transformation.python_transformation import PythonTransformation

from feast.transformation.substrait_transformation import SubstraitTransformation

from feast.utils import _utc_now

from feast.value_type import ValueType

from feast.version_utils import normalize_version_string

warnings.simplefilter("once", DeprecationWarning)

OnDemandSourceType = Union[FeatureView, FeatureViewProjection, RequestSource]

class ODFVErrorMessages:

"""Centralized error message templates for OnDemandFeatureView."""

@staticmethod

def unsupported_source_type(source_type: type, supported_types: str) -> str:

return (

f"Unsupported source type: {source_type.__name__}. "

f"Supported types are {supported_types}."

)

@staticmethod

def singleton_mode_requires_python(current_mode: str) -> str:

return (

f"Singleton mode is only supported with mode='python', "

f"but mode='{current_mode}' was specified. Either disable singleton "

f"(singleton=False) or change mode to 'python'."

)

@staticmethod

def online_store_requires_entities() -> str:

return (

"OnDemandFeatureView configured with write_to_online_store=True "

"must have at least one entity defined. Either add entities or "

"set write_to_online_store=False."

)

@staticmethod

def no_transformation_provided() -> str:

return (

"OnDemandFeatureView must have a valid feature_transformation. "

"Provide either a udf parameter or a feature_transformation parameter."

)

@staticmethod

def duplicate_source_names(overlapping_names: set) -> str:

return (

f"Source names must be unique across all source types. "

f"Found duplicate names: {overlapping_names}"

)

@staticmethod

def no_sources_configured() -> str:

return (

"OnDemandFeatureView must have at least one source. "

"Add either FeatureView/FeatureViewProjection sources or RequestSource sources."

)

@staticmethod

def mode_transformation_mismatch(

mode: str, expected_type: str, actual_type: str

) -> str:

return f"Mode '{mode}' requires {expected_type}, but got {actual_type}."

@staticmethod

def unknown_source_type_in_proto(source_type: str | None) -> str:

return f"Unknown source type in protobuf: {source_type}"

@staticmethod

def unsupported_transformation_type(transformation_type: str) -> str:

return f"Unsupported transformation type: {transformation_type}"

@staticmethod

def backward_compatible_udf_missing() -> str:

return "Backward compatible UDF requires user_defined_function field"

@staticmethod

def unsupported_mode_for_udf(mode: str) -> str:

return f"Unsupported mode '{mode}' for user_defined_function"

@typechecked

class OnDemandFeatureView(BaseFeatureView):

"""

[Experimental] An OnDemandFeatureView defines a logical group of features that are

generated by applying a transformation on a set of input sources, such as feature

views and request data sources.

Attributes:

name: The unique name of the on demand feature view.

features: The list of features in the output of the on demand feature view.

source_feature_view_projections: A map from input source names to actual input

sources with type FeatureViewProjection.

source_request_sources: A map from input source names to the actual input

sources with type RequestSource.

feature_transformation: The user defined transformation.

description: A human-readable description.

tags: A dictionary of key-value pairs to store arbitrary metadata.

owner: The owner of the on demand feature view, typically the email of the primary

maintainer.

org: The organizational unit that owns this on demand feature view (e.g. "ads",

"search"). Defaults to empty string.

"""

_TRACK_METRICS_TAG = "feast:track_metrics"

_INPUT_SCHEMA_SOURCE_PREFIX = "__input_schema__"

name: str

entities: Optional[List[str]]

features: List[Field]

source_feature_view_projections: dict[str, FeatureViewProjection]

source_request_sources: dict[str, RequestSource]

feature_transformation: Optional[Transformation]

mode: str

description: str

tags: dict[str, str]

owner: str

org: str

write_to_online_store: bool

singleton: bool

track_metrics: bool

udf: Optional[FunctionType]

udf_string: Optional[str]

aggregations: List[Aggregation]

enabled: bool

state: FeatureViewState

_raw_feature_transformation_proto: Optional[Any] = None

def __init__( # noqa: C901

self,

*,

name: str,

entities: Optional[List[Entity]] = None,

schema: Optional[List[Field]] = None,

sources: Optional[List[OnDemandSourceType]] = None,

input_schema: Optional[List[Field]] = None,

udf: Optional[FunctionType] = None,

udf_string: Optional[str] = "",

feature_transformation: Optional[Transformation] = None,

mode: str = "pandas",

description: str = "",

tags: Optional[dict[str, str]] = None,

owner: str = "",

org: str = "",

write_to_online_store: bool = False,

singleton: bool = False,

track_metrics: bool = False,

aggregations: Optional[List[Aggregation]] = None,

version: str = "latest",

enabled: bool = True,

):

"""

Creates an OnDemandFeatureView object.

Args:

name: The unique name of the on demand feature view.

entities (optional): The list of names of entities that this feature view is associated with.

schema: The list of features in the output of the on demand feature view, after

the transformation has been applied.

sources: A map from input source names to the actual input sources, which may be

feature views, or request data sources. These sources serve as inputs to the udf,

which will refer to them by name.

input_schema (optional): A list of Fields describing data that is accepted as input

but not stored directly as features — e.g. aggregation columns, normalization

parameters, thresholds, or other contextual values passed at request time.

When provided, sources is not required — an internal RequestSource will be

created automatically.

udf: The user defined transformation function, which must take pandas

dataframes as inputs.

udf_string: The source code version of the udf (for diffing and displaying in Web UI)

feature_transformation: The user defined transformation.

mode: Mode of execution (e.g., Pandas or Python native)

description (optional): A human-readable description.

tags (optional): A dictionary of key-value pairs to store arbitrary metadata.

owner (optional): The owner of the on demand feature view, typically the email

of the primary maintainer.

org (optional): The organizational unit that owns this feature view

(e.g. "ads", "search").

write_to_online_store (optional): A boolean that indicates whether to write the on demand feature view to

the online store for faster retrieval.

singleton (optional): A boolean that indicates whether the transformation is executed on a singleton

(only applicable when mode="python").

track_metrics (optional): Whether to emit Prometheus timing metrics

(``feast_feature_server_transformation_duration_seconds``) for

this ODFV. Defaults to ``False``. Set to ``True`` to opt in

to per-ODFV transformation duration tracking when the server

is started with metrics enabled.

aggregations (optional): List of aggregations to apply before transformation.

"""

super().__init__(

name=name,

features=schema,

description=description,

tags=tags,

owner=owner,

)

self.org = org

self.version = version

schema = schema or []

self.entities = [e.name for e in entities] if entities else [DUMMY_ENTITY_NAME]

self.input_schema = input_schema

self.mode = mode.lower()

self.udf = udf

self.udf_string = udf_string

self.source_feature_view_projections: dict[str, FeatureViewProjection] = {}

self.source_request_sources: dict[str, RequestSource] = {}

self._input_schema_sentinel: Optional[RequestSource] = None

# Strip any existing sentinel from sources (handles __copy__ round-trip)

effective_sources: List[OnDemandSourceType] = [

s

for s in (sources or [])

if not (

isinstance(s, RequestSource)

and s.name.startswith(self._INPUT_SCHEMA_SOURCE_PREFIX)

)

]

if input_schema is not None:

# Automatically create an internal RequestSource from input_schema.

# Stored privately so it does not appear in source_request_sources for

# external consumers (e.g. the feature server, apply(), utils.py).

self._input_schema_sentinel = RequestSource(

name=f"{self._INPUT_SCHEMA_SOURCE_PREFIX}{name}",

schema=input_schema,

)

self.source_request_sources[self._input_schema_sentinel.name] = (

self._input_schema_sentinel

)

elif not effective_sources:

raise ValueError(

"Either 'sources' or 'input_schema' must be provided for OnDemandFeatureView."

)

self.sources = effective_sources

# Process each source with explicit type handling

for odfv_source in effective_sources:

self._add_source_to_collections(odfv_source)

features: List[Field] = []

self.entity_columns = []

join_keys: List[str] = []

if entities:

for entity in entities:

join_keys.append(entity.join_key)

# Ensure that entities have unique join keys.

if len(set(join_keys)) < len(join_keys):

raise ValueError(

"A feature view should not have entities that share a join key."

)

for field in schema:

if field.name in join_keys:

self.entity_columns.append(field)

# Confirm that the inferred type matches the specified entity type, if it exists.

matching_entities = (

[e for e in entities if e.join_key == field.name]

if entities

else []

)

assert len(matching_entities) == 1

entity = matching_entities[0]

if entity.value_type != ValueType.UNKNOWN:

if from_value_type(entity.value_type) != field.dtype:

raise ValueError(

f"Entity {entity.name} has type {entity.value_type}, which does not match the inferred type {field.dtype}."

)

else:

features.append(field)

self.features = features

if feature_transformation is not None:

self.feature_transformation = feature_transformation

elif self.udf is not None:

self.feature_transformation = self.get_feature_transformation()

else:

self.feature_transformation = None

self.write_to_online_store = write_to_online_store

self.singleton = singleton

if self.singleton and self.mode != "python":

raise ValueError(

ODFVErrorMessages.singleton_mode_requires_python(self.mode)

)

self.track_metrics = track_metrics

self.aggregations = aggregations or []

self.enabled = enabled

self.state = FeatureViewState.STATE_UNSPECIFIED

if input_schema is not None and self.aggregations:

input_field_names = {f.name for f in input_schema}

unknown = [

agg.column

for agg in self.aggregations

if agg.column and agg.column not in input_field_names

]

if unknown:

raise ValueError(

f"Aggregation column(s) {unknown} not found in input_schema "

f"for OnDemandFeatureView '{name}'. "

f"Available fields: {sorted(input_field_names)}"

)

def _add_source_to_collections(self, odfv_source: OnDemandSourceType) -> None:

"""

Add a source to the appropriate collection with explicit type checking.

Args:

odfv_source: The source to add (RequestSource, FeatureViewProjection, or FeatureView)

Raises:

ValueError: If the source type is not supported

"""

if isinstance(odfv_source, RequestSource):

self.source_request_sources[odfv_source.name] = odfv_source

elif isinstance(odfv_source, FeatureViewProjection):

self.source_feature_view_projections[odfv_source.name] = odfv_source

elif isinstance(odfv_source, FeatureView):

# FeatureView sources use their projection

self.source_feature_view_projections[odfv_source.name] = (

odfv_source.projection

)

else:

raise ValueError(

ODFVErrorMessages.unsupported_source_type(

type(odfv_source),

"RequestSource, FeatureViewProjection, and FeatureView",

)

)

def get_feature_transformation(self) -> Transformation:

if not self.udf:

raise ValueError(ODFVErrorMessages.no_transformation_provided())

if self.mode in (

TransformationMode.PANDAS,

TransformationMode.PYTHON,

) or self.mode in ("pandas", "python"):

return Transformation(

mode=self.mode, udf=self.udf, udf_string=self.udf_string or ""

)

elif self.mode == TransformationMode.SUBSTRAIT or self.mode == "substrait":

return SubstraitTransformation.from_ibis(self.udf, self.sources)

else:

raise ValueError(

f"Unsupported transformation mode: {self.mode} for OnDemandFeatureView"

)

@property

def proto_class(self) -> type[OnDemandFeatureViewProto]:

return OnDemandFeatureViewProto

def __copy__(self):

fv = OnDemandFeatureView(

name=self.name,

schema=self.features,

sources=list(self.source_feature_view_projections.values())

+ list(self.source_request_sources.values()),

input_schema=self.input_schema,

feature_transformation=self.feature_transformation,

mode=self.mode,

description=self.description,

tags=self.tags,

owner=self.owner,

org=self.org,

write_to_online_store=self.write_to_online_store,

singleton=self.singleton,

version=self.version,

track_metrics=self.track_metrics,

aggregations=self.aggregations,

)

fv.entities = self.entities

fv.features = self.features

fv.projection = copy.copy(self.projection)

fv.entity_columns = copy.copy(self.entity_columns)

fv.enabled = self.enabled

fv.state = self.state

return fv

def _schema_or_udf_changed(self, other: "BaseFeatureView") -> bool:

"""Check for OnDemandFeatureView schema/UDF changes."""

if super()._schema_or_udf_changed(other):

return True

if not isinstance(other, OnDemandFeatureView):

return True

# UDF/transformation changes

# Handle None cases for feature_transformation

if (

self.feature_transformation is None

and other.feature_transformation is not None

):

return True

if (

self.feature_transformation is not None

and other.feature_transformation is None

):

return True

if (

self.feature_transformation is not None

and other.feature_transformation is not None

and self.feature_transformation != other.feature_transformation

):

return True

if self.mode != other.mode:

return True

if (

self.source_feature_view_projections

!= other.source_feature_view_projections

):

return True

if self.source_request_sources != other.source_request_sources:

return True

if sorted(self.entity_columns) != sorted(other.entity_columns):

return True

if self.aggregations != other.aggregations:

return True

# Skip configuration: write_to_online_store, singleton

return False

def __eq__(self, other):

if not isinstance(other, OnDemandFeatureView):

return False

# Note, no longer evaluating the base feature view layer as ODFVs can have

# multiple datasources and a base_feature_view only has one source

# though maybe that shouldn't be true

if (

self.source_feature_view_projections

!= other.source_feature_view_projections

or self.description != other.description

or self.source_request_sources != other.source_request_sources

or self.mode != other.mode

or self.feature_transformation != other.feature_transformation

or self.write_to_online_store != other.write_to_online_store

or sorted(self.entity_columns) != sorted(other.entity_columns)

or self.singleton != other.singleton

or self.track_metrics != other.track_metrics

or self.aggregations != other.aggregations

or normalize_version_string(self.version)

!= normalize_version_string(other.version)

or self.org != other.org

):

return False

return True

@property

def join_keys(self) -> List[str]:

"""Returns a list of all the join keys."""

return [entity.name for entity in self.entity_columns]

@property

def schema(self) -> List[Field]:

return list(set(self.entity_columns + self.features))

def ensure_valid(self):

"""

Validates the state of this feature view locally.

Raises:

ValueError: If the OnDemandFeatureView configuration is invalid.

"""

super().ensure_valid()

# Validate write_to_online_store configuration

self._validate_online_store_config()

# Validate singleton mode configuration

self._validate_singleton_config()

# Validate sources configuration

self._validate_sources_config()

# Validate transformation compatibility

self._validate_transformation_config()

def _validate_online_store_config(self) -> None:

"""Validate write_to_online_store configuration."""

if self.write_to_online_store and not self.entities:

raise ValueError(ODFVErrorMessages.online_store_requires_entities())

def _validate_singleton_config(self) -> None:

"""Validate singleton mode configuration."""

if self.singleton and self.mode != "python":

raise ValueError(

ODFVErrorMessages.singleton_mode_requires_python(self.mode)

)

def _validate_sources_config(self) -> None:

"""Validate sources configuration."""

if not self.source_feature_view_projections and not self.source_request_sources:

raise ValueError(ODFVErrorMessages.no_sources_configured())

# Validate source names are unique

fv_names = set(self.source_feature_view_projections.keys())

req_names = set(self.source_request_sources.keys())

overlapping_names = fv_names.intersection(req_names)

if overlapping_names:

raise ValueError(

ODFVErrorMessages.duplicate_source_names(overlapping_names)

)

def _validate_transformation_config(self) -> None:

"""Validate transformation configuration."""

# Aggregations provide their own transformation; no udf/feature_transformation required.

if self.aggregations:

return

if not self.feature_transformation:

raise ValueError(ODFVErrorMessages.no_transformation_provided())

# Validate mode compatibility with transformation type

if self.mode in ("pandas", "python"):

from feast.transformation.pandas_transformation import PandasTransformation

from feast.transformation.python_transformation import PythonTransformation

expected_types = (PandasTransformation, PythonTransformation)

if not isinstance(self.feature_transformation, expected_types):

raise ValueError(

ODFVErrorMessages.mode_transformation_mismatch(

self.mode,

"PandasTransformation or PythonTransformation",

type(self.feature_transformation).__name__,

)

)

elif self.mode == "substrait":

from feast.transformation.substrait_transformation import (

SubstraitTransformation,

)

if not isinstance(self.feature_transformation, SubstraitTransformation):

raise ValueError(

ODFVErrorMessages.mode_transformation_mismatch(

self.mode,

"SubstraitTransformation",

type(self.feature_transformation).__name__,

)

)

def __hash__(self):

return super().__hash__()

def to_proto(self) -> OnDemandFeatureViewProto:

"""

Converts an on demand feature view object to its protobuf representation.

Returns:

A OnDemandFeatureViewProto protobuf.

"""

meta = OnDemandFeatureViewMeta()

if self.created_timestamp:

meta.created_timestamp.FromDatetime(self.created_timestamp)

if self.last_updated_timestamp:

meta.last_updated_timestamp.FromDatetime(self.last_updated_timestamp)

if self.current_version_number is not None:

meta.current_version_number = self.current_version_number

if self.state != FeatureViewState.STATE_UNSPECIFIED:

meta.state = self.state.to_proto()

sources = {}

for source_name, fv_projection in self.source_feature_view_projections.items():

sources[source_name] = OnDemandSource(

feature_view_projection=fv_projection.to_proto(),

)

for (

source_name,

request_sources,

) in self.source_request_sources.items():

sources[source_name] = OnDemandSource(

request_data_source=request_sources.to_proto()

)

# Serialize the input_schema sentinel so that from_proto() can reconstruct

# input_schema correctly; it is excluded from source_request_sources so that

# external consumers never see it as a real data source.

if self._input_schema_sentinel is not None:

sources[self._input_schema_sentinel.name] = OnDemandSource(

request_data_source=self._input_schema_sentinel.to_proto()

)

if getattr(self, "_raw_feature_transformation_proto", None) is not None:

feature_transformation = self._raw_feature_transformation_proto

else:

feature_transformation = transformation_to_proto(

self.feature_transformation

)

tags = dict(self.tags) if self.tags else {}

if self.track_metrics:

tags[self._TRACK_METRICS_TAG] = "true"

else:

tags.pop(self._TRACK_METRICS_TAG, None)

spec = OnDemandFeatureViewSpec(

name=self.name,

entities=self.entities or None,

entity_columns=[

field.to_proto() for field in self.entity_columns if self.entity_columns

],

features=[feature.to_proto() for feature in self.features],

sources=sources,

feature_transformation=feature_transformation,

mode=self.mode,

description=self.description,

tags=tags,

owner=self.owner,

org=self.org,

write_to_online_store=self.write_to_online_store,

singleton=self.singleton or False,

aggregations=[agg.to_proto() for agg in self.aggregations],

version=self.version,

disabled=not self.enabled,

)

return OnDemandFeatureViewProto(spec=spec, meta=meta)

@classmethod

def from_proto(

cls,

on_demand_feature_view_proto: OnDemandFeatureViewProto,

skip_udf: bool = False,

):

"""

Creates an on demand feature view from a protobuf representation.

Args:

on_demand_feature_view_proto: A protobuf representation of an on-demand feature view.

skip_udf: A boolean indicating whether to skip loading the udf

Returns:

A OnDemandFeatureView object based on the on-demand feature view protobuf.

"""

# Parse sources from proto

sources = cls._parse_sources_from_proto(

on_demand_feature_view_proto, skip_udf=skip_udf

)

# Detect and strip input_schema sentinel from sources

input_schema: Optional[List[Field]] = None

sources_without_sentinel: List[OnDemandSourceType] = []

for source in sources:

if isinstance(source, RequestSource) and source.name.startswith(

cls._INPUT_SCHEMA_SOURCE_PREFIX

):

input_schema = source.schema

else:

sources_without_sentinel.append(source)

sources = sources_without_sentinel

# Parse transformation from proto (skip UDF deserialization if requested)

transformation = cls._parse_transformation_from_proto(

on_demand_feature_view_proto, skip_udf=skip_udf

)

# Parse optional fields with defaults

optional_fields = cls._parse_optional_fields_from_proto(

on_demand_feature_view_proto

)

# Extract track_metrics from proto tags and strip the internal key

# so it doesn't leak into user-facing self.tags.

proto_tags = dict(on_demand_feature_view_proto.spec.tags)

track_metrics = (

proto_tags.pop(cls._TRACK_METRICS_TAG, "false").lower() == "true"

)

# Create the OnDemandFeatureView object

on_demand_feature_view_obj = cls(

name=on_demand_feature_view_proto.spec.name,

schema=cls._parse_features_from_proto(on_demand_feature_view_proto),

sources=cast(List[OnDemandSourceType], sources),

input_schema=input_schema,

feature_transformation=transformation,

mode=on_demand_feature_view_proto.spec.mode or "pandas",

description=on_demand_feature_view_proto.spec.description,

tags=proto_tags,

owner=on_demand_feature_view_proto.spec.owner,

org=on_demand_feature_view_proto.spec.org,

write_to_online_store=optional_fields["write_to_online_store"],

singleton=optional_fields["singleton"],

track_metrics=track_metrics,

aggregations=optional_fields["aggregations"],

)

# Set additional attributes that aren't part of the constructor

on_demand_feature_view_obj.entities = optional_fields["entities"]

on_demand_feature_view_obj.entity_columns = optional_fields["entity_columns"]

on_demand_feature_view_obj.enabled = (

not on_demand_feature_view_proto.spec.disabled

)

# Restore lifecycle state from meta.

on_demand_feature_view_obj.state = FeatureViewState.from_proto(

on_demand_feature_view_proto.meta.state

)

# FeatureViewProjections are not saved in the OnDemandFeatureView proto.

# Create the default projection.

on_demand_feature_view_obj.projection = FeatureViewProjection.from_definition(

on_demand_feature_view_obj

)

# Restore version fields.

spec_version = on_demand_feature_view_proto.spec.version

on_demand_feature_view_obj.version = spec_version or "latest"

cvn = on_demand_feature_view_proto.meta.current_version_number

if cvn > 0:

on_demand_feature_view_obj.current_version_number = cvn

elif cvn == 0 and spec_version and spec_version.lower() != "latest":

on_demand_feature_view_obj.current_version_number = 0

else:

on_demand_feature_view_obj.current_version_number = None

if skip_udf and on_demand_feature_view_proto.spec.HasField(

"feature_transformation"

):

on_demand_feature_view_obj._raw_feature_transformation_proto = (

on_demand_feature_view_proto.spec.feature_transformation

)

# Set timestamps if present

cls._set_timestamps_from_proto(

on_demand_feature_view_proto, on_demand_feature_view_obj

)

return on_demand_feature_view_obj

@classmethod

def _parse_sources_from_proto(

cls, proto: OnDemandFeatureViewProto, skip_udf: bool = False

) -> List[OnDemandSourceType]:

"""Parse and convert sources from the protobuf representation."""

sources: List[OnDemandSourceType] = []

for _, on_demand_source in proto.spec.sources.items():

source_type = on_demand_source.WhichOneof("source")

if source_type == "feature_view":

sources.append(

FeatureView.from_proto(

on_demand_source.feature_view, skip_udf=skip_udf

).projection

)

elif source_type == "feature_view_projection":

sources.append(

FeatureViewProjection.from_proto(

on_demand_source.feature_view_projection

)

)

elif source_type == "request_data_source":

sources.append(

RequestSource.from_proto(on_demand_source.request_data_source)

)

else:

raise ValueError(

ODFVErrorMessages.unknown_source_type_in_proto(source_type)

)

return sources

@classmethod

def _parse_transformation_from_proto(

cls, proto: OnDemandFeatureViewProto, skip_udf: bool = False

) -> Optional[Transformation]:

"""Parse and convert the transformation from the protobuf representation."""

if skip_udf:

return None

feature_transformation = proto.spec.feature_transformation

transformation_type = feature_transformation.WhichOneof("transformation")

mode = proto.spec.mode

if transformation_type == "user_defined_function":

udf_proto = feature_transformation.user_defined_function

# Check for non-empty UDF body

if udf_proto.body_text:

if mode == "pandas":

return PandasTransformation.from_proto(udf_proto)

elif mode == "python":

return PythonTransformation.from_proto(udf_proto)

else:

raise ValueError(ODFVErrorMessages.unsupported_mode_for_udf(mode))

else:

# Handle backward compatibility case with empty body_text

return cls._handle_backward_compatible_udf(proto)

elif transformation_type == "substrait_transformation":

return SubstraitTransformation.from_proto(

feature_transformation.substrait_transformation

)

elif transformation_type is None:

if proto.spec.aggregations:

return None

# Handle backward compatibility case where feature_transformation is cleared

return cls._handle_backward_compatible_udf(proto)

else:

raise ValueError(

ODFVErrorMessages.unsupported_transformation_type(transformation_type)

)

@classmethod

def _handle_backward_compatible_udf(

cls, proto: OnDemandFeatureViewProto

) -> Transformation:

"""Handle backward compatibility for UDFs with empty body_text."""

if not hasattr(proto.spec, "user_defined_function"):

raise ValueError(ODFVErrorMessages.backward_compatible_udf_missing())

old_udf = proto.spec.user_defined_function

backwards_compatible_udf = UserDefinedFunctionProto(

name=old_udf.name,

body=old_udf.body,

body_text=old_udf.body_text,

)

return PandasTransformation.from_proto(

user_defined_function_proto=backwards_compatible_udf,

)

@classmethod

def _parse_features_from_proto(cls, proto: OnDemandFeatureViewProto) -> List[Field]:

"""Parse features from the protobuf representation."""

return [

Field(

name=feature.name,

dtype=from_value_type(ValueType(feature.value_type)),

vector_index=feature.vector_index,

vector_length=feature.vector_length,

vector_search_metric=feature.vector_search_metric,

)

for feature in proto.spec.features

]

@classmethod

def _parse_optional_fields_from_proto(cls, proto: OnDemandFeatureViewProto) -> dict:

"""Parse optional fields from protobuf with appropriate defaults."""

spec = proto.spec

# Parse write_to_online_store

write_to_online_store = False

if hasattr(spec, "write_to_online_store"):

write_to_online_store = spec.write_to_online_store

# Parse entities

entities = []

if hasattr(spec, "entities"):

entities = list(spec.entities)

# Parse entity_columns

entity_columns = []

if hasattr(spec, "entity_columns"):

entity_columns = [

Field.from_proto(field_proto) for field_proto in spec.entity_columns

]

# Parse singleton

singleton = False

if hasattr(spec, "singleton"):

singleton = spec.singleton

# Parse aggregations

aggregations = []

if hasattr(spec, "aggregations"):

aggregations = [

Aggregation.from_proto(aggregation_proto)

for aggregation_proto in spec.aggregations

]

return {

"write_to_online_store": write_to_online_store,

"entities": entities,

"entity_columns": entity_columns,

"singleton": singleton,

"aggregations": aggregations,

}

@classmethod

def _set_timestamps_from_proto(

cls, proto: OnDemandFeatureViewProto, obj: "OnDemandFeatureView"

) -> None:

"""Set timestamp fields on the object if they exist in the proto."""

if proto.meta.HasField("created_timestamp"):

obj.created_timestamp = proto.meta.created_timestamp.ToDatetime()

if proto.meta.HasField("last_updated_timestamp"):

obj.last_updated_timestamp = proto.meta.last_updated_timestamp.ToDatetime()

def get_request_data_schema(self) -> dict[str, ValueType]:

schema: dict[str, ValueType] = {}

for request_source in self.source_request_sources.values():

if isinstance(request_source.schema, list):

new_schema = {}

for field in request_source.schema:

new_schema[field.name] = field.dtype.to_value_type()

schema.update(new_schema)

elif isinstance(request_source.schema, dict):

schema.update(request_source.schema)

else:

raise TypeError(

f"Request source schema is not correct type: ${str(type(request_source.schema))}"

)

# Include fields from the input_schema sentinel (stored privately)

if self._input_schema_sentinel is not None:

for field in self._input_schema_sentinel.schema:

schema[field.name] = field.dtype.to_value_type()

return schema

def _get_projected_feature_name(self, feature: str) -> str:

return f"{self.projection.name_to_use()}__{feature}"

def transform_ibis(

self,

ibis_table,

full_feature_names: bool = False,

):

from ibis.expr.types import Table

if not isinstance(ibis_table, Table):

raise TypeError("transform_ibis only accepts ibis.expr.types.Table")

if not isinstance(self.feature_transformation, SubstraitTransformation):

raise TypeError(

"The feature_transformation is not SubstraitTransformation type while calling transform_ibis()."

)

# Apply common preprocessing to ensure both full and short feature names exist

ibis_table, columns_to_cleanup = self._preprocess_ibis_table(ibis_table)

# Apply the transformation

transformed_table = self.feature_transformation.transform_ibis(ibis_table)

# Clean up temporary columns

if columns_to_cleanup:

transformed_table = transformed_table.drop(*columns_to_cleanup)

# Apply final column renaming based on full_feature_names preference

return self._postprocess_ibis_table(transformed_table, full_feature_names)

def _preprocess_ibis_table(self, ibis_table):

"""

Preprocess ibis table to ensure both full and short feature names exist.

Returns the modified table and columns that need cleanup.

"""

columns_to_cleanup = []

for source_fv_projection in self.source_feature_view_projections.values():

for feature in source_fv_projection.features:

full_feature_ref = f"{source_fv_projection.name}__{feature.name}"

if full_feature_ref in ibis_table.columns:

# Make sure the partial feature name is always present

ibis_table = ibis_table.mutate(

**{feature.name: ibis_table[full_feature_ref]}

)

columns_to_cleanup.append(feature.name)

elif feature.name in ibis_table.columns:

# Make sure the full feature name is always present

ibis_table = ibis_table.mutate(

**{full_feature_ref: ibis_table[feature.name]}

)

columns_to_cleanup.append(full_feature_ref)

return ibis_table, columns_to_cleanup