-
Notifications
You must be signed in to change notification settings - Fork 1.4k
feat: Version-pinning for FeatureService (online + offline) #6718
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
42343fa
80878ac
32c5674
573931b
3083b97
1400657
18239cc
cc4bcce
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| '<feature_view>[@<version>][:<feature>]' 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 | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Docstring says apply/plan resolve these refs, but
Please reject (or resolve) unresolved |
||
| 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( | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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( | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
The docs recommend string refs, but this fails: odfv = on_demand_feature_view(...)(src_fv)
svc = FeatureService(name="svc", features=["my_odfv"])
store.apply([src_fv, odfv, svc]) # FeatureViewNotFoundObject form |
||
| 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( | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This mutates the source FeatureView's own projection.
FeatureView(version="v2")does not setprojection.version_tagitself, so afterFeatureService(features=[fv])the caller'sfv.projection.version_tagbecomes2.Online table naming (
compute_versioned_name) reads that field, so a later materialize/get on the same object can unexpectedly hit*_v2. The string-ref path alreadycopy.copys for this reason — please copy here too before stampingversion_tag.