diff --git a/docs/reference/alpha-feature-view-versioning.md b/docs/reference/alpha-feature-view-versioning.md index 5cdf2845ebc..7c309df44c0 100644 --- a/docs/reference/alpha-feature-view-versioning.md +++ b/docs/reference/alpha-feature-view-versioning.md @@ -42,7 +42,7 @@ feast apply # Detects change → v2 {% hint style="info" %} Version history tracking is **always active** — no configuration needed. Every `feast apply` that changes a feature view automatically records a version snapshot. -To enable **versioned online reads** (e.g., `fv@v2:feature`), add `enable_online_feature_view_versioning: true` to your registry config in `feature_store.yaml`: +To enable **versioned reads** (e.g., `fv@v2:feature`), add `enable_online_feature_view_versioning: true` to your registry config in `feature_store.yaml`: ```yaml registry: @@ -50,7 +50,7 @@ registry: enable_online_feature_view_versioning: true ``` -When this flag is off, version-qualified refs (e.g., `fv@v2:feature`) in online reads will raise errors, but version history, version listing, version pinning, and version lookups all work normally. +Despite the `online` in its name, this flag gates **both** online (`get_online_features`) and offline (`get_historical_features`) versioned resolution — they share the same code path. When it is off, version-qualified refs (e.g., `fv@v2:feature`) raise errors, but version history, version listing, version pinning, and version lookups all work normally. {% endhint %} ## Pinning to a Specific Version @@ -92,6 +92,50 @@ After reverting with a pin, you can go back to normal auto-incrementing behavior | `"v0"`, `"v1"`, `"v2"`, ... | Pin to a specific version number | | `"version0"`, `"version1"`, ... | Equivalent long form (case-insensitive) | +## Pinning a Version in a Feature Service + +A `FeatureService` freezes the exact feature definitions a model was trained and +served on, so it can pin a specific historical feature view version. Two ways to +do this: + +**1. String feature references (recommended).** A `FeatureService` entry may be a +string using the same `[@][:]` syntax as +`get_historical_features`/`get_online_features`. No need to import or reconstruct +the historical feature view object — the pin is resolved against the registry at +`feast apply` time: + +```python +from feast import FeatureService + +pinned_service = FeatureService( + name="model_v1_service", + features=[ + "driver_stats@v0:trips_today", # pinned version, single feature + "driver_stats@v1", # pinned version, whole view + "driver_stats", # latest (unpinned), whole view + ], +) +``` + +**2. A version-pinned feature view object.** Passing a `FeatureView(version="v2")` +object (or a slice of it) into `features` carries its version into the service: + +```python +driver_stats_v2 = FeatureView(name="driver_stats", version="v2", ...) +pinned_service = FeatureService( + name="model_v1_service", + features=[driver_stats_v2[["trips_today"]]], +) +``` + +Either way, the resolved projection renders as `driver_stats@v2` and both online +(`get_online_features`) and offline (`get_historical_features`) retrieval return +the pinned snapshot instead of the promoted version. An unversioned entry +(`"driver_stats"` or a `version="latest"` object) behaves exactly as before. +Requires `enable_online_feature_view_versioning: true` (see +[Configuration](#configuration)); a pinned string ref only resolves once that +version exists in the registry's history. + ## Staged Publishing (`--no-promote`) By default, `feast apply` atomically saves a version snapshot **and** promotes it to the active definition. For breaking schema changes, you may want to stage the new version without disrupting unversioned consumers. @@ -223,7 +267,6 @@ ambiguity, the following characters are reserved and must not appear in feature ## Known Limitations - **Online store coverage** — Version-qualified reads (`@v`) are SQLite-only today. Other online stores are follow-up work. -- **Offline store versioning** — Versioned historical retrieval is not yet supported. +- **Offline store versioning** — Version-qualified historical retrieval (`get_historical_features`) is supported, including through a version-pinned feature service (see [Pinning a Version in a Feature Service](#pinning-a-version-in-a-feature-service)). It requires `enable_online_feature_view_versioning: true`. - **Version deletion** — There is no mechanism to prune old versions from the registry. - **Cross-version joins** — Joining features from different versions of the same feature view in `get_historical_features` is not supported. -- **Feature services** — Feature services always resolve to the active (promoted) version. `--no-promote` versions are not served until promoted. diff --git a/sdk/python/feast/feature_service.py b/sdk/python/feast/feature_service.py index 5cc61372e0f..e53c8f4e3c2 100644 --- a/sdk/python/feast/feature_service.py +++ b/sdk/python/feast/feature_service.py @@ -1,3 +1,4 @@ +import copy from datetime import datetime from typing import TYPE_CHECKING, Dict, List, Optional, Union @@ -24,6 +25,7 @@ from feast.protos.feast.core.FeatureService_pb2 import ( FeatureServiceSpec as FeatureServiceSpecProto, ) +from feast.version_utils import parse_version if TYPE_CHECKING: from feast.infra.registry.base_registry import BaseRegistry @@ -48,7 +50,8 @@ class FeatureService: """ name: str - _features: List[Union[FeatureView, OnDemandFeatureView, LabelView]] + _features: List[Union[FeatureView, OnDemandFeatureView, LabelView, str]] + _pending_feature_refs: List[str] feature_view_projections: List[FeatureViewProjection] description: str tags: Dict[str, str] @@ -63,7 +66,7 @@ def __init__( self, *, name: str, - features: List[Union[FeatureView, OnDemandFeatureView, LabelView]], + features: List[Union[FeatureView, OnDemandFeatureView, LabelView, str]], tags: Optional[Dict[str, str]] = None, description: str = "", owner: str = "", @@ -75,8 +78,18 @@ def __init__( Args: name: The unique name of the feature service. - features: A list containing feature views and feature view - projections, representing the features in the feature service. + features: A list containing feature views, feature view projections, + and/or string feature references, representing the features in + the feature service. A string entry uses the same + '[@][:]' syntax accepted by + ``get_historical_features``/``get_online_features`` — e.g. + "driver_stats" (latest, all features), "driver_stats@v2" + (pinned version, all features), or "driver_stats@v2:trips_today" + (pinned version, single feature). String refs are resolved + against the registry when the feature service is applied + (``FeatureStore.apply``), so a historical version can be pinned + without importing or reconstructing the underlying FeatureView + object. description (optional): A human-readable description. tags (optional): A dictionary of key-value pairs to store arbitrary metadata. owner (optional): The owner of the feature view, typically the email of the @@ -86,6 +99,7 @@ def __init__( """ self.name = name self._features = features + self._pending_feature_refs = [] self.feature_view_projections = [] self.description = description self.tags = tags or {} @@ -95,8 +109,80 @@ def __init__( self.logging_config = logging_config self.precompute_online = precompute_online for feature_grouping in self._features: - if isinstance(feature_grouping, BaseFeatureView): - self.feature_view_projections.append(feature_grouping.projection) + if isinstance(feature_grouping, str): + # No registry at construction time; resolved in resolve_pending_refs. + self._pending_feature_refs.append(feature_grouping) + elif isinstance(feature_grouping, BaseFeatureView): + projection = feature_grouping.projection + # If the source feature view is version-pinned (e.g. + # FeatureView(version="v2")), stamp that version onto the + # projection so name_to_use() renders "fv@v2" and retrieval + # resolves the pinned snapshot. The default version ("latest") + # leaves version_tag as None, preserving existing behavior for + # every unversioned feature service. + fv_version = getattr(feature_grouping, "version", None) + if projection.version_tag is None and fv_version: + is_latest, version_num = parse_version(fv_version) + if not is_latest: + projection.version_tag = version_num + self.feature_view_projections.append(projection) + + def resolve_pending_refs( + self, + project: str, + registry: "BaseRegistry", + fvs_to_update: Optional[Dict[str, Union[FeatureView, BaseFeatureView]]] = None, + ) -> None: + """Resolve string feature refs (see ``__init__``) into projections. + + Called automatically by ``FeatureStore.apply``/``plan`` so the pin is + baked into the applied service instead of re-resolved on every read. + + A version-pinned ref (``"fv@v2"``) always resolves from the registry's + snapshot for that version, never from ``fvs_to_update`` (which only + holds the batch's "latest" objects). An unversioned ref resolves from + ``fvs_to_update`` first, else the promoted version. + + Raises: + ValueError: If a ref names a feature not on the resolved view. + """ + if not self._pending_feature_refs: + return + + from feast.utils import _parse_feature_or_view_ref + + fvs_to_update = fvs_to_update or {} + for ref in self._pending_feature_refs: + fv_name, version_num, feature_name = _parse_feature_or_view_ref(ref) + + if version_num is not None: + feature_view = registry.get_feature_view_by_version( + fv_name, project, version_num, allow_cache=False + ) + elif fv_name in fvs_to_update: + feature_view = fvs_to_update[fv_name] + else: + feature_view = registry.get_any_feature_view( + fv_name, project, allow_cache=False + ) + + # copy so we never mutate the source view's own projection. + projection = copy.copy(feature_view.projection) + if version_num is not None: + projection.version_tag = version_num + + if feature_name is not None: + matches = [f for f in projection.features if f.name == feature_name] + if not matches: + raise ValueError( + f"Invalid feature reference '{ref}': feature " + f"'{feature_name}' not found on feature view '{fv_name}'." + ) + projection.features = matches + + self.feature_view_projections.append(projection) + + self._pending_feature_refs = [] def infer_features( self, fvs_to_update: Dict[str, Union[FeatureView, BaseFeatureView]] @@ -113,6 +199,9 @@ def infer_features( contains all the feature views necessary to run inference. """ for feature_grouping in self._features: + if isinstance(feature_grouping, str): + # Already resolved by resolve_pending_refs before inference. + continue if isinstance(feature_grouping, BaseFeatureView): projection = feature_grouping.projection @@ -213,7 +302,9 @@ def prepare_for_apply( self.infer_features(fvs_to_update=fvs_to_update) return self - resolved_features: List[Union[FeatureView, OnDemandFeatureView, LabelView]] = [] + resolved_features: List[ + Union[FeatureView, OnDemandFeatureView, LabelView, str] + ] = [] for projection in self.feature_view_projections: try: feature_view = registry.get_any_feature_view( diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 546f6f7680d..0e96ef91405 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -1226,6 +1226,11 @@ def _make_inferences( ] } for feature_service in feature_services_to_update: + # Resolve string feature refs (e.g. "driver_stats@v2") before + # inference. No-op for object-only services. + feature_service.resolve_pending_refs( + self.project, self.registry, fvs_to_update=fvs_to_update_map + ) feature_service.infer_features(fvs_to_update=fvs_to_update_map) def _validate_materialize_version( diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py index c9233c77d60..fa86ad1a325 100644 --- a/sdk/python/feast/infra/offline_stores/dask.py +++ b/sdk/python/feast/infra/offline_stores/dask.py @@ -49,6 +49,7 @@ from feast.saved_dataset import SavedDatasetStorage from feast.utils import ( _get_requested_feature_views_to_features_dict, + _get_requested_on_demand_feature_views, compute_non_entity_date_range, ) @@ -197,7 +198,9 @@ def get_historical_features( ) = _get_requested_feature_views_to_features_dict( feature_refs, feature_views, - registry.list_on_demand_feature_views(config.project), + _get_requested_on_demand_feature_views( + feature_refs, config.project, registry + ), ) entity_df_event_timestamp_range = ( diff --git a/sdk/python/feast/infra/offline_stores/offline_utils.py b/sdk/python/feast/infra/offline_stores/offline_utils.py index 7ccaf965c9c..ef6d0584296 100644 --- a/sdk/python/feast/infra/offline_stores/offline_utils.py +++ b/sdk/python/feast/infra/offline_stores/offline_utils.py @@ -21,7 +21,11 @@ from feast.infra.registry.base_registry import BaseRegistry from feast.repo_config import RepoConfig from feast.type_map import feast_value_type_to_pa -from feast.utils import _get_requested_feature_views_to_features_dict, to_naive_utc +from feast.utils import ( + _get_requested_feature_views_to_features_dict, + _get_requested_on_demand_feature_views, + to_naive_utc, +) from feast.value_type import ValueType DEFAULT_ENTITY_DF_EVENT_TIMESTAMP_COL = "event_timestamp" @@ -116,7 +120,9 @@ def get_feature_view_query_context( feature_views_to_feature_map, on_demand_feature_views_to_features, ) = _get_requested_feature_views_to_features_dict( - feature_refs, feature_views, registry.list_on_demand_feature_views(project) + feature_refs, + feature_views, + _get_requested_on_demand_feature_views(feature_refs, project, registry), ) query_context = [] diff --git a/sdk/python/feast/on_demand_feature_view.py b/sdk/python/feast/on_demand_feature_view.py index 3ab188da334..4afb69b9206 100644 --- a/sdk/python/feast/on_demand_feature_view.py +++ b/sdk/python/feast/on_demand_feature_view.py @@ -1345,16 +1345,9 @@ def _get_sample_values_by_type(self) -> dict[ValueType, list[Any]]: 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 + from feast.utils import _get_requested_on_demand_feature_views + + return _get_requested_on_demand_feature_views(feature_refs, project, registry) def on_demand_feature_view( diff --git a/sdk/python/feast/repo_config.py b/sdk/python/feast/repo_config.py index 775a62aab57..940dd757b5e 100644 --- a/sdk/python/feast/repo_config.py +++ b/sdk/python/feast/repo_config.py @@ -197,10 +197,13 @@ class RegistryConfig(FeastBaseModel): only reset the project but not all the projects""" enable_online_feature_view_versioning: StrictBool = False - """ bool: Enable versioned online store tables and version-qualified reads - (e.g., 'fv@v2:feature'). When True, each schema version gets its own - online store table and can be queried independently. Version history - tracking in the registry is always active regardless of this setting. """ + """ bool: Enable versioned resolution of version-qualified feature references + (e.g., 'fv@v2:feature'). Despite the 'online' in its name, this flag gates + both online and offline (get_historical_features) versioned reads, which + share the same resolution path; when True, each schema version also gets + its own online store table and can be queried independently. Version + history tracking in the registry is always active regardless of this + setting. """ mcp: Optional[McpRegistryConfig] = None """ McpRegistryConfig: MCP (Model Context Protocol) configuration for the registry REST server. """ diff --git a/sdk/python/feast/utils.py b/sdk/python/feast/utils.py index 831ed622d06..3bac7c05ccb 100644 --- a/sdk/python/feast/utils.py +++ b/sdk/python/feast/utils.py @@ -63,26 +63,31 @@ USER_AGENT = "{}/{}".format(APPLICATION_NAME, get_version()) -def _parse_feature_ref(ref: str) -> Tuple[str, Optional[int], str]: - """Parse 'fv_name@version:feature' into (fv_name, version_number, feature_name). +def _parse_feature_or_view_ref(ref: str) -> Tuple[str, Optional[int], Optional[str]]: + """Parse 'fv_name[@version][:feature]' into (fv_name, version_number, feature_name). + + Unlike ``_parse_feature_ref``, the ':' suffix is optional, so this + also parses whole-view references (e.g. a FeatureService entry that pins a + version but selects all of the view's features). When no ':' is + present, feature_name is None. When no @version is present, version_number + is None (meaning 'latest'). - If no @version is present, version_number is None (meaning 'latest'). Examples: - 'driver_stats:trips' -> ('driver_stats', None, 'trips') - 'driver_stats@v2:trips' -> ('driver_stats', 2, 'trips') + 'driver_stats:trips' -> ('driver_stats', None, 'trips') + 'driver_stats@v2:trips' -> ('driver_stats', 2, 'trips') 'driver_stats@latest:trips' -> ('driver_stats', None, 'trips') + 'driver_stats' -> ('driver_stats', None, None) + 'driver_stats@v2' -> ('driver_stats', 2, None) """ import re colon_idx = ref.find(":") if colon_idx < 0: - raise ValueError( - f"Invalid feature reference '{ref}'. Expected format: ':' " - f"or '@:'" - ) - - fv_part = ref[:colon_idx] - feature_name = ref[colon_idx + 1 :] + fv_part = ref + feature_name: Optional[str] = None + else: + fv_part = ref[:colon_idx] + feature_name = ref[colon_idx + 1 :] at_idx = fv_part.find("@") if at_idx < 0: @@ -103,6 +108,27 @@ def _parse_feature_ref(ref: str) -> Tuple[str, Optional[int], str]: return (fv_name, int(match.group(1)), feature_name) +def _parse_feature_ref(ref: str) -> Tuple[str, Optional[int], str]: + """Parse 'fv_name@version:feature' into (fv_name, version_number, feature_name). + + The ':' suffix is required; use ``_parse_feature_or_view_ref`` when + a whole-view reference (no feature) should be accepted. + + If no @version is present, version_number is None (meaning 'latest'). + Examples: + 'driver_stats:trips' -> ('driver_stats', None, 'trips') + 'driver_stats@v2:trips' -> ('driver_stats', 2, 'trips') + 'driver_stats@latest:trips' -> ('driver_stats', None, 'trips') + """ + fv_name, version_num, feature_name = _parse_feature_or_view_ref(ref) + if feature_name is None: + raise ValueError( + f"Invalid feature reference '{ref}'. Expected format: ':' " + f"or '@:'" + ) + return (fv_name, version_num, feature_name) + + def _strip_version_from_ref(ref: str) -> str: """Strip @version from a feature reference, returning 'fv_name:feature'. @@ -112,6 +138,67 @@ def _strip_version_from_ref(ref: str) -> str: return f"{fv_name}:{feature_name}" +def _get_requested_on_demand_feature_views( + feature_refs: List[str], + project: str, + registry: "BaseRegistry", + allow_cache: bool = True, +) -> List["OnDemandFeatureView"]: + """Resolve the ODFVs referenced by ``feature_refs``, honouring ``fv@vN`` pins. + + Offline stores re-fetch ODFVs from the registry independently of the + already-resolved online path (``_get_feature_views_to_use``). This mirrors + that version-aware resolution so a pinned ODFV ref resolves to the pinned + snapshot; unversioned refs resolve to the promoted view (unchanged behavior). + """ + from feast.on_demand_feature_view import OnDemandFeatureView + + promoted_by_name = { + odfv.name: odfv + for odfv in registry.list_on_demand_feature_views( + project, allow_cache=allow_cache + ) + } + + # Dedupe refs by (name, version), preserving first-seen order. + requested: Dict[Tuple[str, Optional[int]], None] = {} + for ref in feature_refs: + fv_name, version_num, _ = _parse_feature_ref(ref) + requested[(fv_name, version_num)] = None + + resolved: List["OnDemandFeatureView"] = [] + seen: Set[Tuple[str, Optional[int]]] = set() + for fv_name, version_num in requested: + if version_num is not None: + try: + odfv = registry.get_feature_view_by_version( + fv_name, project, version_num, allow_cache + ) + except NotImplementedError: + # v0 fallback for registries without versioned lookup. + if version_num == 0: + odfv = registry.get_any_feature_view(fv_name, project, allow_cache) + else: + raise + if not isinstance(odfv, OnDemandFeatureView): + continue # ref belongs to a regular FeatureView, not an ODFV + if odfv.projection is not None: + odfv.projection.version_tag = version_num + else: + promoted = promoted_by_name.get(fv_name) + if promoted is None: + continue + odfv = promoted + + key = (odfv.name, version_num) + if key in seen: + continue + seen.add(key) + resolved.append(odfv) + + return resolved + + def get_user_agent(): return USER_AGENT @@ -1261,7 +1348,7 @@ def _get_feature_views_to_use( if isinstance(features, FeatureService): feature_views = [ - (projection.name, None, projection) + (projection.name, projection.version_tag, projection) for projection in features.feature_view_projections ] else: diff --git a/sdk/python/tests/unit/test_feature_service_versioning.py b/sdk/python/tests/unit/test_feature_service_versioning.py new file mode 100644 index 00000000000..f1444eee3c8 --- /dev/null +++ b/sdk/python/tests/unit/test_feature_service_versioning.py @@ -0,0 +1,223 @@ +from __future__ import annotations + +import pytest + +from feast.errors import FeastObjectNotFoundException +from feast.feature_service import FeatureService +from feast.feature_view import FeatureView +from feast.field import Field +from feast.infra.offline_stores.file_source import FileSource +from feast.types import Float32, String + + +def _build_feature_view( + version: str = "latest", + schema: list[Field] | None = None, +) -> FeatureView: + file_source = FileSource(name="my-file-source", path="test.parquet") + return FeatureView( + name="driver_stats", + entities=[], + schema=schema + or [ + Field(name="conv_rate", dtype=Float32), + Field(name="acc_rate", dtype=String), + ], + source=file_source, + version=version, + ) + + +class _MockRegistry: + """Registry stub for resolve_pending_refs: version snapshots + promoted view.""" + + def __init__(self, by_version=None, promoted=None): + self._by_version = by_version or {} # (name, version) -> FeatureView + self._promoted = promoted or {} # name -> FeatureView + self.calls: list[tuple] = [] + + def get_feature_view_by_version( + self, name, project, version_number, allow_cache=False + ): + self.calls.append(("by_version", name, version_number)) + key = (name, version_number) + if key not in self._by_version: + raise FeastObjectNotFoundException(f"{name}@v{version_number}") + return self._by_version[key] + + def get_any_feature_view(self, name, project, allow_cache=False): + self.calls.append(("any", name)) + if name not in self._promoted: + raise FeastObjectNotFoundException(name) + return self._promoted[name] + + +def test_pinned_feature_view_stamps_version_tag_on_projection(): + fv = _build_feature_view(version="v2") + service = FeatureService(name="pinned_service", features=[fv]) + + projection = service.feature_view_projections[0] + assert projection.version_tag == 2 + assert projection.name_to_use() == "driver_stats@v2" + + +def test_pinned_feature_view_slice_stamps_version_tag(): + fv = _build_feature_view(version="v2") + service = FeatureService(name="pinned_service", features=[fv[["conv_rate"]]]) + + projection = service.feature_view_projections[0] + assert projection.version_tag == 2 + assert projection.name_to_use() == "driver_stats@v2" + assert [f.name for f in projection.features] == ["conv_rate"] + + +def test_version_tag_survives_proto_round_trip(): + fv = _build_feature_view(version="v3") + service = FeatureService(name="pinned_service", features=[fv]) + + restored = FeatureService.from_proto(service.to_proto()) + + projection = restored.feature_view_projections[0] + assert projection.version_tag == 3 + assert projection.name_to_use() == "driver_stats@v3" + + +def test_unversioned_feature_view_leaves_version_tag_none(): + # Backward compatibility: the default "latest" version must not stamp a + # version_tag, so existing unversioned feature services are unaffected. + fv = _build_feature_view(version="latest") + service = FeatureService(name="unpinned_service", features=[fv]) + + projection = service.feature_view_projections[0] + assert projection.version_tag is None + assert projection.name_to_use() == "driver_stats" + + +# --- string feature refs (resolve_pending_refs) --- + + +def test_string_refs_stashed_not_resolved_at_construction(): + service = FeatureService(name="svc", features=["driver_stats@v1"]) + assert service.feature_view_projections == [] + assert service._pending_feature_refs == ["driver_stats@v1"] + + +def test_string_ref_pins_version_and_selects_single_feature(): + fv_v1 = _build_feature_view(version="v1") + registry = _MockRegistry(by_version={("driver_stats", 1): fv_v1}) + service = FeatureService(name="svc", features=["driver_stats@v1:conv_rate"]) + + service.resolve_pending_refs("proj", registry) + + projection = service.feature_view_projections[0] + assert projection.version_tag == 1 + assert projection.name_to_use() == "driver_stats@v1" + assert [f.name for f in projection.features] == ["conv_rate"] + assert ("by_version", "driver_stats", 1) in registry.calls + # Source view's own projection must not be mutated. + assert [f.name for f in fv_v1.projection.features] == ["conv_rate", "acc_rate"] + + +def test_string_ref_pins_version_whole_view(): + fv_v1 = _build_feature_view(version="v1") + registry = _MockRegistry(by_version={("driver_stats", 1): fv_v1}) + service = FeatureService(name="svc", features=["driver_stats@v1"]) + + service.resolve_pending_refs("proj", registry) + + projection = service.feature_view_projections[0] + assert projection.version_tag == 1 + assert [f.name for f in projection.features] == ["conv_rate", "acc_rate"] + + +def test_unversioned_string_ref_uses_promoted_view(): + promoted = _build_feature_view(version="latest") + registry = _MockRegistry(promoted={"driver_stats": promoted}) + service = FeatureService(name="svc", features=["driver_stats"]) + + service.resolve_pending_refs("proj", registry) + + projection = service.feature_view_projections[0] + assert projection.version_tag is None + assert projection.name_to_use() == "driver_stats" + assert registry.calls == [("any", "driver_stats")] + + +def test_unversioned_string_ref_prefers_fvs_to_update(): + batch_fv = _build_feature_view(version="latest") + registry = _MockRegistry() + service = FeatureService(name="svc", features=["driver_stats"]) + + service.resolve_pending_refs( + "proj", registry, fvs_to_update={"driver_stats": batch_fv} + ) + + assert service.feature_view_projections[0].version_tag is None + # Resolved from the apply batch, so the registry was never queried. + assert registry.calls == [] + + +def test_pinned_ref_ignores_fvs_to_update(): + # The pinned v1 snapshot has a different schema than the "latest" object in + # the apply batch; the pin must resolve from the registry, not the batch. + fv_v1 = _build_feature_view( + version="v1", + schema=[ + Field(name="conv_rate", dtype=Float32), + Field(name="acc_rate", dtype=String), + ], + ) + latest = _build_feature_view( + version="latest", + schema=[ + Field(name="conv_rate", dtype=Float32), + Field(name="new_col", dtype=String), + ], + ) + registry = _MockRegistry(by_version={("driver_stats", 1): fv_v1}) + service = FeatureService(name="svc", features=["driver_stats@v1"]) + + service.resolve_pending_refs( + "proj", registry, fvs_to_update={"driver_stats": latest} + ) + + projection = service.feature_view_projections[0] + assert projection.version_tag == 1 + assert [f.name for f in projection.features] == ["conv_rate", "acc_rate"] + assert ("by_version", "driver_stats", 1) in registry.calls + + +def test_string_ref_unknown_feature_raises(): + fv_v1 = _build_feature_view(version="v1") + registry = _MockRegistry(by_version={("driver_stats", 1): fv_v1}) + service = FeatureService(name="svc", features=["driver_stats@v1:missing"]) + + with pytest.raises(ValueError, match="'missing' not found"): + service.resolve_pending_refs("proj", registry) + + +def test_string_ref_unknown_version_raises(): + registry = _MockRegistry() # no versions registered + service = FeatureService(name="svc", features=["driver_stats@v9"]) + + with pytest.raises(FeastObjectNotFoundException): + service.resolve_pending_refs("proj", registry) + + +def test_resolve_pending_refs_is_noop_without_refs(): + fv = _build_feature_view(version="latest") + service = FeatureService(name="svc", features=[fv]) + service.resolve_pending_refs("proj", _MockRegistry()) + # Only the object-based projection is present; no extra work done. + assert len(service.feature_view_projections) == 1 + + +def test_infer_features_skips_string_refs(): + fv_v1 = _build_feature_view(version="v1") + registry = _MockRegistry(by_version={("driver_stats", 1): fv_v1}) + service = FeatureService(name="svc", features=["driver_stats@v1"]) + service.resolve_pending_refs("proj", registry) + + # Must not raise "invalid type str" for the still-present string entry. + service.infer_features(fvs_to_update={}) + assert service.feature_view_projections[0].version_tag == 1 diff --git a/sdk/python/tests/unit/test_odfv_offline_versioning.py b/sdk/python/tests/unit/test_odfv_offline_versioning.py new file mode 100644 index 00000000000..0341a7eb30d --- /dev/null +++ b/sdk/python/tests/unit/test_odfv_offline_versioning.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +import pandas as pd +import pytest + +from feast.feature_view import FeatureView +from feast.field import Field +from feast.infra.offline_stores.file_source import FileSource +from feast.on_demand_feature_view import ( + OnDemandFeatureView, + PandasTransformation, +) +from feast.types import Float32 +from feast.utils import _get_requested_on_demand_feature_views + + +def _udf(features_df: pd.DataFrame) -> pd.DataFrame: + df = pd.DataFrame() + df["output1"] = features_df["feature1"] + return df + + +def _build_odfv(name: str = "my_odfv") -> OnDemandFeatureView: + source = FeatureView( + name="source_fv", + entities=[], + schema=[Field(name="feature1", dtype=Float32)], + source=FileSource(name="src", path="test.parquet"), + ) + return OnDemandFeatureView( + name=name, + sources=[source], + schema=[Field(name="output1", dtype=Float32)], + feature_transformation=PandasTransformation(udf=_udf, udf_string="src"), + ) + + +def _build_feature_view(name: str = "plain_fv") -> FeatureView: + return FeatureView( + name=name, + entities=[], + schema=[Field(name="feature1", dtype=Float32)], + source=FileSource(name="src2", path="test.parquet"), + ) + + +class _MockRegistry: + def __init__(self, promoted=None, by_version=None, versioned_supported=True): + self._promoted = promoted or [] + self._by_version = by_version or {} # (name, version) -> view + self._versioned_supported = versioned_supported + + def list_on_demand_feature_views(self, project, allow_cache=True): + return self._promoted + + def get_feature_view_by_version( + self, name, project, version_number, allow_cache=False + ): + if not self._versioned_supported: + raise NotImplementedError + return self._by_version[(name, version_number)] + + def get_any_feature_view(self, name, project, allow_cache=False): + for odfv in self._promoted: + if odfv.name == name: + return odfv + raise KeyError(name) + + +def test_unversioned_ref_returns_promoted_odfv(): + promoted = _build_odfv("my_odfv") + registry = _MockRegistry(promoted=[promoted]) + + result = _get_requested_on_demand_feature_views( + ["my_odfv:output1"], "proj", registry + ) + + assert [odfv.name for odfv in result] == ["my_odfv"] + assert result[0].projection.version_tag is None + + +def test_versioned_ref_returns_pinned_snapshot_with_version_tag(): + pinned = _build_odfv("my_odfv") + registry = _MockRegistry(by_version={("my_odfv", 1): pinned}) + + result = _get_requested_on_demand_feature_views( + ["my_odfv@v1:output1"], "proj", registry + ) + + assert len(result) == 1 + assert result[0].projection.version_tag == 1 + assert result[0].projection.name_to_use() == "my_odfv@v1" + + +def test_versioned_ref_to_regular_feature_view_is_skipped(): + # get_feature_view_by_version can return a plain FeatureView; it is not an + # ODFV and must be dropped from the ODFV list. + plain = _build_feature_view("plain_fv") + registry = _MockRegistry(by_version={("plain_fv", 1): plain}) + + result = _get_requested_on_demand_feature_views( + ["plain_fv@v1:feature1"], "proj", registry + ) + + assert result == [] + + +def test_duplicate_refs_are_deduplicated(): + pinned = _build_odfv("my_odfv") + registry = _MockRegistry(by_version={("my_odfv", 1): pinned}) + + result = _get_requested_on_demand_feature_views( + ["my_odfv@v1:output1", "my_odfv@v1:output1"], "proj", registry + ) + + assert len(result) == 1 + + +def test_v0_falls_back_when_versioned_lookup_unsupported(): + promoted = _build_odfv("my_odfv") + registry = _MockRegistry(promoted=[promoted], versioned_supported=False) + + result = _get_requested_on_demand_feature_views( + ["my_odfv@v0:output1"], "proj", registry + ) + + assert [odfv.name for odfv in result] == ["my_odfv"] + + +def test_nonzero_version_reraises_when_lookup_unsupported(): + registry = _MockRegistry(versioned_supported=False) + + with pytest.raises(NotImplementedError): + _get_requested_on_demand_feature_views(["my_odfv@v2:output1"], "proj", registry)