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): raise TypeError( "Comparisons should only involve OnDemandFeatureView class objects." ) # 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 def _postprocess_ibis_table(self, transformed_table, full_feature_names: bool): """ Apply final column renaming to match the desired naming convention. """ rename_columns: dict[str, str] = {} for feature in self.features: short_name = feature.name long_name = self._get_projected_feature_name(feature.name) if short_name in transformed_table.columns and full_feature_names: rename_columns[short_name] = long_name elif long_name in transformed_table.columns and not full_feature_names: rename_columns[long_name] = short_name # Apply renamings for rename_from, rename_to in rename_columns.items(): if rename_from in transformed_table.columns: transformed_table = transformed_table.rename(**{rename_to: rename_from}) return transformed_table def transform_arrow( self, pa_table: pyarrow.Table, full_feature_names: bool = False, ) -> pyarrow.Table: if not isinstance(pa_table, pyarrow.Table): raise TypeError("transform_arrow only accepts pyarrow.Table") # Apply common preprocessing to ensure both full and short feature names exist pa_table, columns_to_cleanup = self._preprocess_arrow_table(pa_table) # Apply the transformation assert self.feature_transformation is not None transformed_table = self.feature_transformation.transform_arrow( pa_table, self.features ) # Clean up temporary columns and apply final renaming return self._postprocess_arrow_table( transformed_table, columns_to_cleanup, full_feature_names ) def _preprocess_arrow_table(self, pa_table: pyarrow.Table): """ Preprocess pyarrow 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 pa_table.column_names: # Make sure the partial feature name is always present pa_table = pa_table.append_column( feature.name, pa_table[full_feature_ref] ) columns_to_cleanup.append(feature.name) elif feature.name in pa_table.column_names: # Make sure the full feature name is always present pa_table = pa_table.append_column( full_feature_ref, pa_table[feature.name] ) columns_to_cleanup.append(full_feature_ref) return pa_table, columns_to_cleanup def _postprocess_arrow_table( self, transformed_table: pyarrow.Table, columns_to_cleanup: list[str], full_feature_names: bool, ) -> pyarrow.Table: """ Clean up temporary columns and apply final column renaming. """ # Determine final column names rename_columns: dict[str, str] = {} for feature in self.features: short_name = feature.name long_name = self._get_projected_feature_name(feature.name) if short_name in transformed_table.column_names and full_feature_names: rename_columns[short_name] = long_name elif long_name in transformed_table.column_names and not full_feature_names: rename_columns[long_name] = short_name # Clean up temporary columns for col in columns_to_cleanup: if col in transformed_table.column_names: transformed_table = transformed_table.drop(col) # Apply column renaming final_column_names = [ rename_columns.get(c, c) for c in transformed_table.column_names ] return transformed_table.rename_columns(final_column_names) def transform_dict( self, feature_dict: dict[str, Any], # type: ignore ) -> dict[str, Any]: """ Transform a dictionary of features using the configured transformation. Handles both singleton and batch transformations. Args: feature_dict: Dictionary containing input features Returns: Dictionary with transformed features """ # Preprocess to ensure both full and short feature names exist preprocessed_dict, columns_to_cleanup = self._preprocess_feature_dict( feature_dict ) # Apply the appropriate transformation based on mode assert self.feature_transformation is not None if self.singleton and self.mode == "python": output_dict = self.feature_transformation.transform_singleton( preprocessed_dict ) else: output_dict = self.feature_transformation.transform(preprocessed_dict) # Clean up temporary columns for feature_name in columns_to_cleanup: if feature_name in output_dict: del output_dict[feature_name] return output_dict def _preprocess_feature_dict( self, feature_dict: dict[str, Any] ) -> tuple[dict[str, Any], list[str]]: """ Preprocess feature dictionary to ensure both full and short feature names exist. Returns the modified dictionary and columns that need cleanup. """ # Create a copy to avoid modifying the original preprocessed_dict = feature_dict.copy() 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 feature_dict: # Make sure the partial feature name is always present preprocessed_dict[feature.name] = feature_dict[full_feature_ref] columns_to_cleanup.append(feature.name) elif feature.name in feature_dict: # Make sure the full feature name is always present preprocessed_dict[full_feature_ref] = feature_dict[feature.name] columns_to_cleanup.append(full_feature_ref) return preprocessed_dict, columns_to_cleanup def infer_features(self) -> None: if self.aggregations and not self.feature_transformation: if not self.features: raise RegistryInferenceFailure( "OnDemandFeatureView", f"Could not infer Features for the feature view '{self.name}'.", ) return assert self.feature_transformation is not None random_input = self._construct_random_input(singleton=self.singleton) inferred_features = self.feature_transformation.infer_features( random_input=random_input, singleton=self.singleton ) if self.features: missing_features = [] for specified_feature in self.features: if not self._feature_exists_in_inferred( specified_feature, inferred_features ): missing_features.append(specified_feature) if missing_features: raise SpecifiedFeaturesNotPresentError( missing_features, inferred_features, self.name ) else: self.features = inferred_features if not self.features: raise RegistryInferenceFailure( "OnDemandFeatureView", f"Could not infer Features for the feature view '{self.name}'.", ) def _feature_exists_in_inferred( self, specified_feature: Field, inferred_features: List[Field] ) -> bool: """ Check if a specified feature exists in the inferred features list. Handles both regular features and array types properly. Args: specified_feature: The feature to check for inferred_features: List of inferred features Returns: True if the feature exists in the inferred features, False otherwise """ # Check for exact feature match first if specified_feature in inferred_features: return True # For array types, we need to check by name since array types # might have different representations between specified and inferred if self._is_array_type(specified_feature.dtype): inferred_feature_names = {f.name for f in inferred_features} return specified_feature.name in inferred_feature_names return False def _is_array_type(self, dtype) -> bool: """Check if the dtype represents an array type.""" # Use proper type checking instead of string comparison dtype_str = str(dtype) return "Array" in dtype_str or "List" in dtype_str or "Set" in dtype_str def _construct_random_input( self, singleton: bool = False ) -> dict[str, Union[list[Any], Any]]: """ Construct random input data for feature inference. Args: singleton: Whether to use singleton values (single values instead of lists) Returns: Dictionary with random sample data for each source feature """ # Get sample values for each ValueType sample_values = self._get_sample_values_by_type() # Convert to singleton values if needed if singleton: sample_values = {k: v[0] for k, v in sample_values.items()} # Default value for missing types default_value = None if not singleton else [None] feature_dict = {} # Add feature view projection features for feature_view_projection in self.source_feature_view_projections.values(): for feature in feature_view_projection.features: value_type = feature.dtype.to_value_type() sample_value = sample_values.get(value_type, default_value) # Add both full and short feature references feature_dict[f"{feature_view_projection.name}__{feature.name}"] = ( sample_value ) feature_dict[feature.name] = sample_value # Add request source features for request_data in self.source_request_sources.values(): for field in request_data.schema: value_type = field.dtype.to_value_type() sample_value = sample_values.get(value_type, default_value) feature_dict[field.name] = sample_value # Add input_schema fields (stored privately outside source_request_sources) if self._input_schema_sentinel is not None: for field in self._input_schema_sentinel.schema: value_type = field.dtype.to_value_type() sample_value = sample_values.get(value_type, default_value) feature_dict[field.name] = sample_value return feature_dict def _get_sample_values_by_type(self) -> dict[ValueType, list[Any]]: """ Get sample values for each supported ValueType. Centralizes the mapping between ValueTypes and their sample values. Returns: Dictionary mapping ValueType to sample values """ # Sample PDF bytes for testing pdf_sample = ( b"%PDF-1.3\n3 0 obj\n<>\nendobj\n" b"4 0 obj\n<>\nstream\nx\x9c\x15\xcc1\x0e\x820\x18@\xe1\x9dS\xbcM]jk$\xd5\xd5(\x83!\x86\xa1\x17\xf8\xa3\xa5`LIh+\xd7W\xc6\xf7\r\xef\xc0\xbd\xd2\xaa\xb6,\xd5\xc5\xb1o\x0c\xa6VZ\xe3znn%\xf3o\xab\xb1\xe7\xa3:Y\xdc\x8bm\xeb\xf3&1\xc8\xd7\xd3\x97\xc82\xe6\x81\x87\xe42\xcb\x87Vb(\x12<\xdd<=}Jc\x0cL\x91\xee\xda$\xb5\xc3\xbd\xd7\xe9\x0f\x8d\x97 $\nendstream\nendobj\n" b"1 0 obj\n<>\nendobj\n" b"5 0 obj\n<>\nendobj\n" b"2 0 obj\n<<\n/ProcSet [/PDF /Text /ImageB /ImageC /ImageI]\n/Font <<\n/F1 5 0 R\n>>\n/XObject <<\n>>\n>>\nendobj\n" b"6 0 obj\n<<\n/Producer (PyFPDF 1.7.2 http://pyfpdf.googlecode.com/)\n/Title (This is a sample title.)\n/Author (Francisco Javier Arceo)\n/CreationDate (D:20250312165548)\n>>\nendobj\n" b"7 0 obj\n<<\n/Type /Catalog\n/Pages 1 0 R\n/OpenAction [3 0 R /FitH null]\n/PageLayout /OneColumn\n>>\nendobj\n" b"xref\n0 8\n0000000000 65535 f \n0000000272 00000 n \n0000000455 00000 n \n0000000009 00000 n \n0000000087 00000 n \n0000000359 00000 n \n0000000559 00000 n \n0000000734 00000 n \n" b"trailer\n<<\n/Size 8\n/Root 7 0 R\n/Info 6 0 R\n>>\nstartxref\n837\n%%EOF\n" ) # Sample image bytes (minimal JPEG) image_sample = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x01\x00H\x00H\x00\x00\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t\x08\n\x0c\x14\r\x0c\x0b\x0b\x0c\x19\x12\x13\x0f\x14\x1d\x1a\x1f\x1e\x1d\x1a\x1c\x1c $.' \",#\x1c\x1c(7),01444\x1f'9=82<.342\xff\xc0\x00\x11\x08\x00\x01\x00\x01\x01\x01\x11\x00\x02\x11\x01\x03\x11\x01\xff\xc4\x00\x14\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x08\xff\xc4\x00\x14\x10\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xda\x00\x0c\x03\x01\x00\x02\x11\x03\x11\x00\x3f\x00\xaa\xff\xd9" return { # Basic types ValueType.BYTES: [b"hello world"], ValueType.STRING: ["hello world"], ValueType.INT32: [1], ValueType.INT64: [1], ValueType.DOUBLE: [1.0], ValueType.FLOAT: [1.0], ValueType.BOOL: [True], ValueType.UNIX_TIMESTAMP: [_utc_now()], # Special binary types ValueType.PDF_BYTES: [pdf_sample], ValueType.IMAGE_BYTES: [image_sample], # UUID types ValueType.UUID: [uuid.uuid4()], ValueType.TIME_UUID: [uuid.uuid1()], # List types ValueType.BYTES_LIST: [[b"hello world"]], ValueType.STRING_LIST: [["hello world"]], ValueType.INT32_LIST: [[1]], ValueType.INT64_LIST: [[1]], ValueType.DOUBLE_LIST: [[1.0]], ValueType.FLOAT_LIST: [[1.0]], ValueType.BOOL_LIST: [[True]], ValueType.UNIX_TIMESTAMP_LIST: [[_utc_now()]], ValueType.UUID_LIST: [[uuid.uuid4(), uuid.uuid4()]], ValueType.TIME_UUID_LIST: [[uuid.uuid1(), uuid.uuid1()]], # Set types ValueType.BYTES_SET: [{b"hello world", b"foo bar"}], ValueType.STRING_SET: [{"hello world", "foo bar"}], ValueType.INT32_SET: [{1, 2}], ValueType.INT64_SET: [{1, 2}], ValueType.DOUBLE_SET: [{1.0, 2.0}], ValueType.FLOAT_SET: [{1.0, 2.0}], ValueType.BOOL_SET: [{True, False}], ValueType.UNIX_TIMESTAMP_SET: [{_utc_now()}], ValueType.UUID_SET: [{uuid.uuid4(), uuid.uuid4()}], ValueType.TIME_UUID_SET: [{uuid.uuid1(), uuid.uuid1()}], } @staticmethod def get_requested_odfvs( feature_refs, project, registry ) -> list["OnDemandFeatureView"]: all_on_demand_feature_views = registry.list_on_demand_feature_views( project, allow_cache=True ) requested_on_demand_feature_views: list[OnDemandFeatureView] = [] for odfv in all_on_demand_feature_views: for feature in odfv.features: if f"{odfv.name}:{feature.name}" in feature_refs: requested_on_demand_feature_views.append(odfv) break return requested_on_demand_feature_views def on_demand_feature_view( *, name: Optional[str] = None, entities: Optional[List[Entity]] = None, schema: list[Field], sources: Optional[ list[ Union[ FeatureView, RequestSource, FeatureViewProjection, ] ] ] = None, input_schema: Optional[list[Field]] = None, aggregations: Optional[List[Aggregation]] = 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, explode: bool = False, version: str = "latest", ): """ Creates an OnDemandFeatureView object with the given user function as udf. Args: name (optional): The name of the on demand feature view. If not provided, the name will be the name of the user function. 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. mode: The 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 on demand feature view (e.g. "ads", "search"). Defaults to empty string. 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 for this ODFV. Defaults to False. Set to True to opt in when the server is started with metrics. explode (optional): A boolean that indicates whether the transformation explodes the input data into multiple rows. """ def mainify(obj) -> None: # Needed to allow dill to properly serialize the udf. Otherwise, clients will need to have a file with the same # name as the original file defining the ODFV. if obj.__module__ != "__main__": obj.__module__ = "__main__" def decorator(user_function): udf_string = dill.source.getsource(user_function) mainify(user_function) on_demand_feature_view_obj = OnDemandFeatureView( name=name if name is not None else user_function.__name__, sources=sources, input_schema=input_schema, schema=schema, mode=mode, description=description, tags=tags, owner=owner, org=org, write_to_online_store=write_to_online_store, entities=entities, singleton=singleton, track_metrics=track_metrics, aggregations=aggregations, udf=user_function, udf_string=udf_string, version=version, ) functools.update_wrapper( wrapper=on_demand_feature_view_obj, wrapped=user_function ) return on_demand_feature_view_obj return decorator def _empty_odfv_udf_fn(x: Any) -> Any: # just an identity mapping, otherwise we risk tripping some downstream tests return x