From d5ed6ca006827f4555affdcbf98161b597984d6c Mon Sep 17 00:00:00 2001 From: ntkathole Date: Mon, 1 Jun 2026 22:02:48 +0530 Subject: [PATCH] feat: Pre-compute feature service Signed-off-by: ntkathole --- .../concepts/feature-retrieval.md | 25 + .../online-server-performance-tuning.md | 44 +- protos/feast/core/FeatureService.proto | 4 + .../feast/core/PrecomputedFeatureVector.proto | 31 + sdk/python/feast/feature_service.py | 23 +- sdk/python/feast/feature_store.py | 256 +++++ .../feast/infra/online_stores/online_store.py | 491 ++++++++- sdk/python/feast/infra/online_stores/redis.py | 204 +--- .../protos/feast/core/DataSource_pb2.pyi | 717 +++++++------- .../protos/feast/core/FeatureService_pb2.py | 52 +- .../protos/feast/core/FeatureService_pb2.pyi | 8 +- .../core/PrecomputedFeatureVector_pb2.py | 31 + .../core/PrecomputedFeatureVector_pb2.pyi | 76 ++ .../core/PrecomputedFeatureVector_pb2_grpc.py | 4 + .../unit/test_precomputed_feature_vectors.py | 933 ++++++++++++++++++ 15 files changed, 2248 insertions(+), 651 deletions(-) create mode 100644 protos/feast/core/PrecomputedFeatureVector.proto create mode 100644 sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.py create mode 100644 sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.pyi create mode 100644 sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2_grpc.py create mode 100644 sdk/python/tests/unit/test_precomputed_feature_vectors.py diff --git a/docs/getting-started/concepts/feature-retrieval.md b/docs/getting-started/concepts/feature-retrieval.md index b3c4062edfd..7c4dec08ef9 100644 --- a/docs/getting-started/concepts/feature-retrieval.md +++ b/docs/getting-started/concepts/feature-retrieval.md @@ -52,6 +52,31 @@ Applying a feature service does not result in an actual service being deployed. Feature services enable referencing all or some features from a feature view. +#### Pre-computed feature vectors (`precompute_online`) + +For latency-critical online serving, you can enable **pre-computed feature vectors** on a feature service. When `precompute_online=True`, Feast stores all of the service's features for each entity as a single serialized blob in the online store. At read time, this reduces the number of store reads from O(N feature views) to O(1), regardless of how many feature views the service spans. + +```python +# A feature service with pre-computed vectors enabled +low_latency_service = FeatureService( + name="low_latency_inference", + features=[driver_stats_fv, vehicle_stats_fv, route_features_fv], + precompute_online=True, +) +``` + +After running `feast apply`, the pre-computed vectors are automatically built and refreshed whenever you run `feast materialize` or `feast materialize-incremental`. Feast detects which feature services have `precompute_online=True` and rebuilds their vectors for every affected entity after the per-feature-view writes complete. Vectors are also refreshed automatically on `feast push`. + +{% hint style="info" %} +`precompute_online` is **opt-in** — it defaults to `False`. When enabled, the pre-computed path is used exclusively for that service; there is no silent fallback to per-feature-view reads. If vectors are missing or stale, the server raises an error, making problems visible immediately. +{% endhint %} + +{% hint style="warning" %} +`precompute_online` is not compatible with on-demand feature views (ODFVs) that have `write_to_online_store=False`. ODFVs with `write_to_online_store=True` are supported since their values are materialized. +{% endhint %} + +See the [performance tuning guide](../../how-to-guides/online-server-performance-tuning.md#pre-computed-feature-vectors) for benchmarks and detailed configuration. + Retrieving from the online store with a feature service ```python diff --git a/docs/how-to-guides/online-server-performance-tuning.md b/docs/how-to-guides/online-server-performance-tuning.md index 34508ff6ce4..b280fe53cff 100644 --- a/docs/how-to-guides/online-server-performance-tuning.md +++ b/docs/how-to-guides/online-server-performance-tuning.md @@ -35,10 +35,12 @@ When the server processes a `get_online_features()` call, it groups the requeste **Redis exception:** The Redis online store overrides `get_online_features()` to batch all `HMGET` commands across every feature view into a **single pipeline execution**. Because all feature views for the same entity share one Redis hash key, the number of Redis round trips is always **1**, regardless of how many feature views the request touches. This means the "fewer feature views" guideline is less critical for Redis than for other stores — but consolidating feature views still reduces serialization and protobuf overhead at the application layer. {% endhint %} -### Feature services are free +### Feature services are free (and can be faster) A [Feature Service](../getting-started/concepts/feature-retrieval.md) is a named collection of feature references — it's a convenience grouping, not a separate storage or execution unit. Using a feature service adds only a registry lookup (cached) compared to listing features individually. There is no performance penalty for using feature services, and they are the recommended way to define stable, versioned feature sets for production models. +For latency-critical services, feature services also unlock **pre-computed feature vectors** (`precompute_online=True`), which reduce store reads from O(N feature views) to O(1). See the [Pre-computed feature vectors](#pre-computed-feature-vectors) section below for details and benchmarks. + ### ODFV overhead is additive Regular feature views incur only **store read** cost. On-demand feature views add **CPU-bound transformation** cost on top: @@ -83,6 +85,45 @@ Requesting just `combined_score` triggers reads from **both** `driver_stats_fv` --- +## Pre-computed feature vectors + +When a `get_online_features()` request touches multiple feature views, the server issues a separate store read per feature view. For services spanning 5–15+ feature views, this fan-out dominates latency — even with Redis pipeline batching, the protobuf deserialization and response-building overhead grows linearly with the number of views. + +**Pre-computed feature vectors** solve this by storing all of a feature service's features for each entity as a single serialized blob. At read time, the server fetches one blob per entity instead of N reads per feature view, reducing the operation to O(1). + +### How it works + +1. **Define** a feature service with `precompute_online=True`: + +```python +benchmark_service = FeatureService( + name="benchmark_customer_service", + features=[ + customer_demographics_fv, + customer_behavioral_profile, + transaction_7d_aggregations, + transaction_30d_aggregations, + transaction_90d_patterns, + atm_usage_30d, + ], + precompute_online=True, +) +``` + +2. **Apply** the feature service: `feast apply` +3. **Materialize** as usual — vectors are built automatically: `feast materialize-incremental $(date -u +"%Y-%m-%dT%H:%M:%S")` +4. **Read** features as usual — the server automatically uses the pre-computed path: + +```python +features = store.get_online_features( + features=store.get_feature_service("benchmark_customer_service"), + entity_rows=[{"customer_id": "CUST_000001"}], + full_feature_names=True, +) +``` + +--- + ## Worker and connection tuning The Python feature server uses Gunicorn with async workers. Tuning workers, connections, and timeouts directly impacts throughput and tail latency. @@ -1086,4 +1127,5 @@ Reset `skip_dedup` to `false` (or remove it) after the bulk reload. Under normal - [PostgreSQL Online Store](../reference/online-stores/postgres.md) — Connection pooling and SSL configuration - [Redis Online Store](../reference/online-stores/redis.md) — Cluster mode, Sentinel, TTL configuration, and batched reads - [On Demand Feature Views](../reference/beta-on-demand-feature-view.md) — Transformation modes and write-time transforms +- [Feature Services & `precompute_online`](../getting-started/concepts/feature-retrieval.md#pre-computed-feature-vectors-precompute_online) — Concept docs for pre-computed feature vectors - [feature_store.yaml reference](../reference/feature-repository/feature-store-yaml.md) — Full configuration reference including `materialization` options diff --git a/protos/feast/core/FeatureService.proto b/protos/feast/core/FeatureService.proto index 380b2dc3718..557782fe552 100644 --- a/protos/feast/core/FeatureService.proto +++ b/protos/feast/core/FeatureService.proto @@ -38,6 +38,10 @@ message FeatureServiceSpec { // (optional) if provided logging will be enabled for this feature service. LoggingConfig logging_config = 7; + + // When true, a pre-computed feature vector is maintained per entity for this + // service, enabling single-read online retrieval instead of per-feature-view reads. + bool precompute_online = 8; } diff --git a/protos/feast/core/PrecomputedFeatureVector.proto b/protos/feast/core/PrecomputedFeatureVector.proto new file mode 100644 index 00000000000..798df0b8d1e --- /dev/null +++ b/protos/feast/core/PrecomputedFeatureVector.proto @@ -0,0 +1,31 @@ +syntax = "proto3"; +package feast.core; + +option go_package = "github.com/feast-dev/feast/go/protos/feast/core"; +option java_outer_classname = "PrecomputedFeatureVectorProto"; +option java_package = "feast.proto.core"; + +import "google/protobuf/timestamp.proto"; +import "feast/types/Value.proto"; + +// A pre-computed feature vector stores all features for a FeatureService +// as a single serialized blob per entity, enabling O(1) online retrieval. +message PrecomputedFeatureVector { + // Fully-qualified feature names in deterministic order, e.g. "fv1__feat1". + repeated string feature_names = 1; + + // Feature values in the same order as feature_names. + repeated feast.types.Value values = 2; + + // Per-feature-view event timestamps for TTL enforcement at read time. + repeated FeatureViewTimestamp fv_timestamps = 3; + + // Wall-clock time when this vector was assembled. + google.protobuf.Timestamp precomputed_at = 4; +} + +// Event timestamp associated with a specific feature view within the vector. +message FeatureViewTimestamp { + string feature_view_name = 1; + google.protobuf.Timestamp event_timestamp = 2; +} diff --git a/sdk/python/feast/feature_service.py b/sdk/python/feast/feature_service.py index 5c062f15c67..b937d210d6e 100644 --- a/sdk/python/feast/feature_service.py +++ b/sdk/python/feast/feature_service.py @@ -49,6 +49,8 @@ class FeatureService: last_updated_timestamp: Optional[datetime] = None logging_config: Optional[LoggingConfig] = None + precompute_online: bool = False + def __init__( self, *, @@ -58,18 +60,21 @@ def __init__( description: str = "", owner: str = "", logging_config: Optional[LoggingConfig] = None, + precompute_online: bool = False, ): """ Creates a FeatureService object. Args: name: The unique name of the feature service. - feature_view_projections: A list containing feature views and feature view + features: A list containing feature views and feature view projections, representing the features in the feature service. 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 primary maintainer. + precompute_online (optional): When True, a pre-computed feature vector is + maintained per entity for single-read online retrieval. """ self.name = name self._features = features @@ -80,6 +85,7 @@ def __init__( self.created_timestamp = None self.last_updated_timestamp = None 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) @@ -175,6 +181,7 @@ def __eq__(self, other): or self.description != other.description or self.tags != other.tags or self.owner != other.owner + or self.precompute_online != other.precompute_online ): return False @@ -202,6 +209,7 @@ def from_proto(cls, feature_service_proto: FeatureServiceProto): logging_config=LoggingConfig.from_proto( feature_service_proto.spec.logging_config ), + precompute_online=feature_service_proto.spec.precompute_online, ) fs.feature_view_projections.extend( [ @@ -245,9 +253,20 @@ def to_proto(self) -> FeatureServiceProto: logging_config=self.logging_config.to_proto() if self.logging_config else None, + precompute_online=self.precompute_online, ) return FeatureServiceProto(spec=spec, meta=meta) def validate(self): - pass + if not self.precompute_online: + return + + for fv in self._features: + if isinstance(fv, OnDemandFeatureView) and not fv.write_to_online_store: + raise ValueError( + f"FeatureService '{self.name}' has precompute_online=True but " + f"contains OnDemandFeatureView '{fv.name}' with " + f"write_to_online_store=False. On-demand transforms computed at " + f"serve time cannot be pre-computed." + ) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 765e22898b0..a41a70149d1 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -107,6 +107,11 @@ from feast.utils import _get_feature_view_vector_field_metadata, _utc_now from feast.version_utils import parse_version +try: + from datetime import timezone as _timezone +except ImportError: + _timezone = None # type: ignore[assignment,misc] + _track_materialization = None # Lazy-loaded on first materialization call _track_materialization_loaded = False @@ -2151,6 +2156,14 @@ def tqdm_builder(length): end_date, ) + materialized_fv_names = [ + fv.name + for fv in feature_views_to_materialize + if not isinstance(fv, OnDemandFeatureView) + ] + if materialized_fv_names: + self._precompute_affected_services(materialized_fv_names) + # Emit OpenLineage COMPLETE event self._emit_openlineage_materialize_complete( ol_run_id, feature_views_to_materialize @@ -2319,6 +2332,14 @@ def tqdm_builder(length): end_date, ) + materialized_fv_names = [ + fv.name + for fv in feature_views_to_materialize + if not isinstance(fv, OnDemandFeatureView) + ] + if materialized_fv_names: + self._precompute_affected_services(materialized_fv_names) + # Emit OpenLineage COMPLETE event self._emit_openlineage_materialize_complete( ol_run_id, feature_views_to_materialize @@ -2434,6 +2455,236 @@ def _fvs_for_push_source_or_raise( return fvs_with_push_sources + def precompute_feature_service( + self, + feature_service_name: Optional[str] = None, + batch_size: int = 1000, + ) -> int: + """Pre-compute feature vectors for one or all FeatureServices. + + For each FeatureService with ``precompute_online=True`` (or matching + *feature_service_name*), reads every entity's features from the online + store via :meth:`OnlineStore.online_read` and writes a single serialized + blob per entity via :meth:`OnlineStore.write_precomputed_vector`. + + Works with **all** online store backends (Redis, DynamoDB, PostgreSQL, etc.). + + Returns the total number of entity vectors written. + """ + from feast.protos.feast.core.PrecomputedFeatureVector_pb2 import ( + FeatureViewTimestamp, + PrecomputedFeatureVector, + ) + + provider = self._get_provider() + online_store = provider.online_store + + services = self.registry.list_feature_services(self.project) + if feature_service_name: + services = [s for s in services if s.name == feature_service_name] + + total_written = 0 + for svc in services: + if not svc.precompute_online and not feature_service_name: + continue + + fv_projections = svc.feature_view_projections + feature_views = [] + for proj in fv_projections: + fv = self.registry.get_any_feature_view( + proj.name, self.project, allow_cache=True + ) + feature_views.append((fv, proj)) + + if not feature_views: + continue + + feature_names: List[str] = [] + for _fv, proj in feature_views: + fv_name = proj.name_to_use() + for f in proj.features: + feature_names.append(f"{fv_name}__{f.name}") + + # Collect all unique entity key protos from the primary feature view. + # Use online_read to discover entities that exist in the store. + primary_fv = feature_views[0][0] + + # Read entity keys by scanning the primary FV's online data. + # We use get_online_features with the full FeatureService to read + # all features for each entity in a single call, then build vectors. + # + # For stores that support native scanning (like Redis), we try the + # native scan; for others, we use the materialized entity list. + entity_key_protos: List[EntityKey] = [] + try: + from feast.infra.online_stores.redis import RedisOnlineStore + + if isinstance(online_store, RedisOnlineStore): + from feast.infra.key_encoding_utils import ( + deserialize_entity_key, + ) + from feast.infra.online_stores.helpers import _redis_key_prefix + + join_keys = ( + list(primary_fv.join_keys) + if hasattr(primary_fv, "join_keys") + else [] + ) # type: ignore[union-attr] + client = online_store._get_client( # type: ignore[attr-defined] + self.config.online_store + ) + scan_prefix = _redis_key_prefix(join_keys) + project_bytes = self.config.project.encode("utf-8") + seen_keys: set = set() + raw_keys: List[bytes] = [] + for key in client.scan_iter( + b"".join([scan_prefix, b"*", project_bytes]) + ): + if key not in seen_keys: + seen_keys.add(key) + raw_keys.append(key) + proj_len = len(project_bytes) + for rk in raw_keys: + try: + ek = deserialize_entity_key( + rk[:-proj_len], + self.config.entity_key_serialization_version, + ) + if set(ek.join_keys) == set(join_keys): + entity_key_protos.append(ek) + except (ValueError, Exception): + continue + else: + raise NotImplementedError + except (ImportError, NotImplementedError): + _logger.warning( + "Entity scanning not supported for '%s' — " + "precompute requires a store that supports entity scanning", + type(online_store).__name__, + ) + continue + + if not entity_key_protos: + continue + + for batch_start in range(0, len(entity_key_protos), batch_size): + batch_keys = entity_key_protos[batch_start : batch_start + batch_size] + + # Read features for each FV via base class online_read. + fv_data: Dict[ + str, + List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]], + ] = {} + for fv_obj, proj in feature_views: + req_features = [f.name for f in proj.features] + assert isinstance(fv_obj, FeatureView) + rows = online_store.online_read( + config=self.config, + table=fv_obj, + entity_keys=batch_keys, + requested_features=req_features, + ) + fv_data[proj.name_to_use()] = rows + + for entity_idx, entity_key in enumerate(batch_keys): + values: List[ValueProto] = [] + fv_timestamps: list = [] + + for _fv, proj in feature_views: + fv_name = proj.name_to_use() + fv_row_ts, feat_dict = fv_data[fv_name][entity_idx] + + if fv_row_ts: + fv_ts = Timestamp() + fv_ts.FromDatetime(utils.make_tzaware(fv_row_ts)) + fv_timestamps.append( + FeatureViewTimestamp( + feature_view_name=fv_name, + event_timestamp=fv_ts, + ) + ) + + for f in proj.features: + if feat_dict and f.name in feat_dict: + values.append(feat_dict[f.name]) + else: + values.append(ValueProto()) + + now_ts = Timestamp() + now_ts.FromDatetime(datetime.now(tz=_timezone.utc)) + + vector = PrecomputedFeatureVector( + feature_names=feature_names, + values=values, + fv_timestamps=fv_timestamps, + precomputed_at=now_ts, + ) + + online_store.write_precomputed_vector( + config=self.config, + feature_service_name=svc.name, + project=self.config.project, + entity_key=entity_key, + vector_bytes=vector.SerializeToString(), + ) + total_written += 1 + + _logger.info( + "Pre-computed %d entity vectors for FeatureService '%s'", + total_written, + svc.name, + ) + + return total_written + + def _precompute_affected_services(self, materialized_fv_names: List[str]) -> None: + """Trigger precomputation for services affected by materialized FVs.""" + try: + services = self.registry.list_feature_services(self.project) + except Exception: + return + + for svc in services: + if not svc.precompute_online: + continue + svc_fv_names = {p.name for p in svc.feature_view_projections} + if svc_fv_names & set(materialized_fv_names): + try: + self.precompute_feature_service(svc.name) + except Exception: + _logger.warning( + "Failed to precompute vectors for service '%s'", + svc.name, + exc_info=True, + ) + + def _precompute_for_push(self, feature_view_name: str, df: "pd.DataFrame") -> None: + """Re-compute pre-computed vectors for entities affected by a push.""" + try: + services = self.registry.list_feature_services(self.project) + except Exception: + return + + affected = [ + svc + for svc in services + if svc.precompute_online + and any(p.name == feature_view_name for p in svc.feature_view_projections) + ] + + if not affected: + return + + for svc in affected: + try: + self.precompute_feature_service(svc.name) + except Exception: + _logger.warning( + "Failed to precompute vectors for service '%s' after push", + svc.name, + exc_info=True, + ) + def push( self, push_source_name: str, @@ -2452,6 +2703,7 @@ def push( to: Whether to push to online or offline store. Defaults to online store only. transform_on_write: Whether to transform the data before pushing. """ + pushed_fv_names = [] for fv in self._fvs_for_push_source_or_raise( push_source_name, allow_registry_cache ): @@ -2462,11 +2714,15 @@ def push( allow_registry_cache=allow_registry_cache, transform_on_write=transform_on_write, ) + pushed_fv_names.append(fv.name) if to == PushMode.OFFLINE or to == PushMode.ONLINE_AND_OFFLINE: self.write_to_offline_store( fv.name, df, allow_registry_cache=allow_registry_cache ) + if pushed_fv_names: + self._precompute_for_push(pushed_fv_names[0], df) + async def push_async( self, push_source_name: str, diff --git a/sdk/python/feast/infra/online_stores/online_store.py b/sdk/python/feast/infra/online_stores/online_store.py index 49533cbf226..87feba3f830 100644 --- a/sdk/python/feast/infra/online_stores/online_store.py +++ b/sdk/python/feast/infra/online_stores/online_store.py @@ -12,10 +12,14 @@ # See the License for the specific language governing permissions and # limitations under the License. import asyncio +import logging +import time as _time_mod from abc import ABC, abstractmethod from datetime import datetime from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Union +from google.protobuf.timestamp_pb2 import Timestamp + from feast import Entity, utils from feast.batch_feature_view import BatchFeatureView from feast.errors import VersionedOnlineReadNotSupported @@ -33,6 +37,8 @@ from feast.stream_feature_view import StreamFeatureView from feast.value_type import ValueType +logger = logging.getLogger(__name__) + class OnlineStore(ABC): """ @@ -203,32 +209,53 @@ def get_online_features( _read_start = _time.monotonic() - for table, requested_features in grouped_refs: - # Get the correct set of entity values with the correct join keys. - table_entity_values, idxs, output_len = utils._get_unique_entities( - table, - join_key_values, - entity_name_to_join_key_map, - ) - - entity_key_protos = utils._get_entity_key_protos(table_entity_values) + use_precomputed = ( + isinstance(features, FeatureService) + and getattr(features, "precompute_online", False) + and not requested_on_demand_feature_views + ) - # Fetch data for Entities. - read_rows = self.online_read( - config=config, - table=table, - entity_keys=entity_key_protos, - requested_features=requested_features, + precomputed_ok = False + if use_precomputed and grouped_refs: + assert isinstance(features, FeatureService) + first_table = grouped_refs[0][0] + first_entity_values, first_idxs, output_len = utils._get_unique_entities( + first_table, join_key_values, entity_name_to_join_key_map ) + entity_key_protos = utils._get_entity_key_protos(first_entity_values) - utils._populate_response_from_feature_data( - requested_features, - read_rows, - idxs, + blobs = self.read_precomputed_vectors( + config, features.name, project, entity_key_protos + ) + expected_names = self._compute_expected_feature_names( + grouped_refs, full_feature_names + ) + precomputed_ok = self._try_precomputed_fast_path( + blobs, + expected_names, online_features_response, full_feature_names, - table, output_len, + grouped_refs, + registry, + project, + ) + if not precomputed_ok: + raise RuntimeError( + f"FeatureService '{features.name}' has precompute_online=True " + f"but pre-computed vectors could not be read. " + f"Run `feast precompute {features.name}` or materialize to " + f"populate vectors." + ) + + if not precomputed_ok: + self._read_features_per_fv( + config, + grouped_refs, + join_key_values, + entity_name_to_join_key_map, + online_features_response, + full_feature_names, include_feature_view_version_metadata, ) @@ -297,6 +324,117 @@ def _is_versioned_read_supported(self) -> bool: pass return isinstance(self, tuple(supported_types)) + def _read_features_per_fv( + self, + config: RepoConfig, + grouped_refs: List, + join_key_values: Dict, + entity_name_to_join_key_map: Dict, + online_features_response, + full_feature_names: bool, + include_feature_view_version_metadata: bool, + ) -> None: + """Read features one feature-view at a time (generic path). + + Subclasses may override to batch reads more efficiently (e.g. Redis + pipeline). + """ + for table, requested_features in grouped_refs: + table_entity_values, idxs, output_len = utils._get_unique_entities( + table, join_key_values, entity_name_to_join_key_map + ) + entity_key_protos = utils._get_entity_key_protos(table_entity_values) + + read_rows = self.online_read( + config=config, + table=table, + entity_keys=entity_key_protos, + requested_features=requested_features, + ) + + utils._populate_response_from_feature_data( + requested_features, + read_rows, + idxs, + online_features_response, + full_feature_names, + table, + output_len, + include_feature_view_version_metadata, + ) + + async def _read_features_per_fv_async( + self, + config: RepoConfig, + grouped_refs: List, + join_key_values: Dict, + entity_name_to_join_key_map: Dict, + online_features_response, + full_feature_names: bool, + include_feature_view_version_metadata: bool, + ) -> None: + """Async version of :meth:`_read_features_per_fv`. + + Reads all feature views concurrently via ``asyncio.gather``. + Subclasses may override to batch reads more efficiently. + """ + + async def query_table(table, requested_features): + table_entity_values, idxs, output_len = utils._get_unique_entities( + table, join_key_values, entity_name_to_join_key_map + ) + entity_key_protos = utils._get_entity_key_protos(table_entity_values) + read_rows = await self.online_read_async( + config=config, + table=table, + entity_keys=entity_key_protos, + requested_features=requested_features, + ) + return idxs, read_rows, output_len + + all_responses = await asyncio.gather( + *[ + query_table(table, requested_features) + for table, requested_features in grouped_refs + ] + ) + + for (idxs, read_rows, output_len), (table, requested_features) in zip( + all_responses, grouped_refs + ): + utils._populate_response_from_feature_data( + requested_features, + read_rows, + idxs, + online_features_response, + full_feature_names, + table, + output_len, + include_feature_view_version_metadata, + ) + + async def read_precomputed_vectors_async( + self, + config: RepoConfig, + feature_service_name: str, + project: str, + entity_keys: List[EntityKeyProto], + ) -> List[Optional[bytes]]: + """Async version of :meth:`read_precomputed_vectors`. + + The default implementation delegates to the sync method via the event + loop executor. Online stores with native async support should override. + """ + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, + self.read_precomputed_vectors, + config, + feature_service_name, + project, + entity_keys, + ) + async def get_online_features_async( self, config: RepoConfig, @@ -343,26 +481,6 @@ async def get_online_features_async( # Check for versioned reads on unsupported stores self._check_versioned_read_support(grouped_refs) - async def query_table(table, requested_features): - # Get the correct set of entity values with the correct join keys. - table_entity_values, idxs, output_len = utils._get_unique_entities( - table, - join_key_values, - entity_name_to_join_key_map, - ) - - entity_key_protos = utils._get_entity_key_protos(table_entity_values) - - # Fetch data for Entities. - read_rows = await self.online_read_async( - config=config, - table=table, - entity_keys=entity_key_protos, - requested_features=requested_features, - ) - - return idxs, read_rows, output_len - _track_read = False try: from feast.metrics import _config as _metrics_config @@ -376,24 +494,53 @@ async def query_table(table, requested_features): _read_start = _time.monotonic() - all_responses = await asyncio.gather( - *[ - query_table(table, requested_features) - for table, requested_features in grouped_refs - ] + use_precomputed = ( + isinstance(features, FeatureService) + and getattr(features, "precompute_online", False) + and not requested_on_demand_feature_views ) - for (idxs, read_rows, output_len), (table, requested_features) in zip( - all_responses, grouped_refs - ): - utils._populate_response_from_feature_data( - requested_features, - read_rows, - idxs, + precomputed_ok = False + if use_precomputed and grouped_refs: + assert isinstance(features, FeatureService) + first_table = grouped_refs[0][0] + first_entity_values, first_idxs, output_len = utils._get_unique_entities( + first_table, join_key_values, entity_name_to_join_key_map + ) + entity_key_protos = utils._get_entity_key_protos(first_entity_values) + + blobs = await self.read_precomputed_vectors_async( + config, features.name, project, entity_key_protos + ) + expected_names = self._compute_expected_feature_names( + grouped_refs, full_feature_names + ) + precomputed_ok = self._try_precomputed_fast_path( + blobs, + expected_names, online_features_response, full_feature_names, - table, output_len, + grouped_refs, + registry, + project, + ) + if not precomputed_ok: + raise RuntimeError( + f"FeatureService '{features.name}' has precompute_online=True " + f"but pre-computed vectors could not be read. " + f"Run `feast precompute {features.name}` or materialize to " + f"populate vectors." + ) + + if not precomputed_ok: + await self._read_features_per_fv_async( + config, + grouped_refs, + join_key_values, + entity_name_to_join_key_map, + online_features_response, + full_feature_names, include_feature_view_version_metadata, ) @@ -437,6 +584,244 @@ def _build_feature_types( feature_types[f"{table_name}__{field.name}"] = vtype return feature_types + @staticmethod + def _compute_expected_feature_names( + grouped_refs: List, full_feature_names: bool + ) -> List[str]: + """Derive the deterministic feature name list for schema comparison.""" + names: List[str] = [] + for table, requested_features in grouped_refs: + table_name = table.projection.name_to_use() + for fn in requested_features: + if fn.startswith("_ts:"): + continue + names.append(f"{table_name}__{fn}" if full_feature_names else fn) + return names + + @staticmethod + def _try_precomputed_fast_path( + blobs: List[Optional[bytes]], + expected_feature_names: List[str], + online_features_response: Any, + full_feature_names: bool, + num_rows: int, + grouped_refs: List, + registry: BaseRegistry, + project: str, + ) -> bool: + """Build the response from pre-computed vectors. + + Returns True if the fast path succeeded for ALL entities, False otherwise + (caller should fall back to per-FV reads). + """ + from feast.protos.feast.core.PrecomputedFeatureVector_pb2 import ( + PrecomputedFeatureVector, + ) + from feast.protos.feast.serving.ServingService_pb2 import ( + FieldStatus, + GetOnlineFeaturesResponse, + ) + + for i, blob in enumerate(blobs): + if blob is None: + logger.warning( + "Pre-computed vector blob is None for entity index %d", i + ) + return False + + n_features = len(expected_feature_names) + expected_set = set(expected_feature_names) + + # Pre-compute feature-index-to-FV-name mapping once (not per entity). + feat_fv_names: List[Optional[str]] = [] + for fname in expected_feature_names: + feat_fv_names.append(fname.split("__", 1)[0] if "__" in fname else None) + + # Pre-compute FV TTLs once. + fv_ttls: Dict[str, Optional[int]] = {} + for table, _ in grouped_refs: + ttl = table.ttl + fv_ttls[table.projection.name_to_use()] = ( + int(ttl.total_seconds()) if ttl else None + ) + + # Check if any FV actually has a TTL — skip TTL logic entirely if not. + any_ttl = any(v is not None for v in fv_ttls.values()) + + # Parse all blobs, validate schema, build reorder map. + # Schema is typically identical for all entities, so validate against the + # first and then just verify the rest match the first (not the expected list). + first_stored_names: Optional[List[str]] = None + reorder_map: Optional[List[int]] = None + vectors: List[PrecomputedFeatureVector] = [] + + for blob in blobs: + vec = PrecomputedFeatureVector() + vec.ParseFromString(blob) # type: ignore[arg-type] + vectors.append(vec) + + if first_stored_names is None: + first_stored_names = list(vec.feature_names) + if set(first_stored_names) != expected_set: + logger.warning( + "Pre-computed vector schema mismatch: stored=%s, expected=%s", + first_stored_names, + expected_feature_names, + ) + return False + if first_stored_names != expected_feature_names: + name_to_idx = {n: i for i, n in enumerate(first_stored_names)} + reorder_map = [name_to_idx[n] for n in expected_feature_names] + else: + if ( + len(vec.feature_names) != n_features + or list(vec.feature_names) != first_stored_names + ): + logger.warning( + "Pre-computed vector schema varies across entities at index %d", + len(vectors) - 1, + ) + return False + + PRESENT = FieldStatus.PRESENT + OUTSIDE_MAX_AGE = FieldStatus.OUTSIDE_MAX_AGE + null_value = ValueProto() + null_ts = Timestamp() + + feat_values = [[null_value] * num_rows for _ in range(n_features)] + feat_statuses = [[FieldStatus.NOT_FOUND] * num_rows for _ in range(n_features)] + ts_list = [null_ts] * num_rows + + now_secs = _time_mod.time() if any_ttl else 0.0 + + for row_idx, vec in enumerate(vectors): + ts_list[row_idx] = vec.precomputed_at + + # Build per-FV expiry flags once per entity (not per feature). + fv_expired: Optional[Dict[str, bool]] = None + if any_ttl: + fv_ts_map: Dict[str, Timestamp] = { + fvt.feature_view_name: fvt.event_timestamp + for fvt in vec.fv_timestamps + } + fv_expired = {} + for fv_name, ttl_val in fv_ttls.items(): + if ttl_val is not None: + event_ts = fv_ts_map.get(fv_name) + if event_ts: + event_secs = event_ts.seconds + event_ts.nanos / 1e9 + fv_expired[fv_name] = (now_secs - event_secs) > ttl_val + else: + fv_expired[fv_name] = False + else: + fv_expired[fv_name] = False + + stored_values = vec.values + for out_idx in range(n_features): + src_idx = reorder_map[out_idx] if reorder_map else out_idx + feat_values[out_idx][row_idx] = stored_values[src_idx] + + if fv_expired: + feat_fv = feat_fv_names[out_idx] + if feat_fv and fv_expired.get(feat_fv, False): + feat_statuses[out_idx][row_idx] = OUTSIDE_MAX_AGE + continue + feat_statuses[out_idx][row_idx] = PRESENT + + online_features_response.metadata.feature_names.val.extend( + expected_feature_names + ) + for f_idx in range(n_features): + online_features_response.results.append( + GetOnlineFeaturesResponse.FeatureVector( + values=feat_values[f_idx], + statuses=feat_statuses[f_idx], + event_timestamps=ts_list, + ) + ) + return True + + def write_precomputed_vector( + self, + config: RepoConfig, + feature_service_name: str, + project: str, + entity_key: EntityKeyProto, + vector_bytes: bytes, + ) -> None: + """Write a pre-computed feature vector blob for a single entity. + + Stores the blob as a ``bytes_val`` feature under a synthetic FeatureView + named ``__precomputed__{service_name}``, using the generic + ``online_write_batch`` API so this works for every online store without + store-specific overrides. + """ + from feast.field import Field + from feast.types import Bytes + + synthetic_fv = FeatureView( + name=f"__precomputed__{feature_service_name}", + entities=[], + schema=[Field(name="vector", dtype=Bytes)], + source=None, + ) + val = ValueProto(bytes_val=vector_bytes) + ts = datetime.utcnow() + self.online_write_batch( + config=config, + table=synthetic_fv, + data=[(entity_key, {"vector": val}, ts, None)], + progress=None, + ) + + def read_precomputed_vectors( + self, + config: RepoConfig, + feature_service_name: str, + project: str, + entity_keys: List[EntityKeyProto], + ) -> List[Optional[bytes]]: + """Read pre-computed feature vector blobs for a batch of entities. + + Returns a list aligned with *entity_keys*. Each element is either the + serialized ``PrecomputedFeatureVector`` bytes or ``None`` when no + pre-computed vector exists for that entity. + + Reads from the synthetic FeatureView written by + :meth:`write_precomputed_vector` using the generic ``online_read`` API, + so this works for every online store without store-specific overrides. + """ + from feast.field import Field + from feast.types import Bytes + + synthetic_fv = FeatureView( + name=f"__precomputed__{feature_service_name}", + entities=[], + schema=[Field(name="vector", dtype=Bytes)], + source=None, + ) + try: + rows = self.online_read( + config=config, + table=synthetic_fv, + entity_keys=entity_keys, + requested_features=["vector"], + ) + except Exception: + return [None] * len(entity_keys) + + result: List[Optional[bytes]] = [] + for _ts, feature_dict in rows: + if feature_dict and "vector" in feature_dict: + val = feature_dict["vector"] + if val.HasField("bytes_val"): + result.append(val.bytes_val) + else: + result.append(None) + else: + result.append(None) + return result + @abstractmethod def update( self, diff --git a/sdk/python/feast/infra/online_stores/redis.py b/sdk/python/feast/infra/online_stores/redis.py index 262df0c863d..ad9e378a95f 100644 --- a/sdk/python/feast/infra/online_stores/redis.py +++ b/sdk/python/feast/infra/online_stores/redis.py @@ -22,7 +22,6 @@ Dict, List, Literal, - Mapping, Optional, Sequence, Tuple, @@ -33,7 +32,6 @@ from pydantic import StrictStr from feast import Entity, FeatureView, RepoConfig, utils -from feast.feature_service import FeatureService from feast.infra.key_encoding_utils import serialize_entity_key from feast.infra.online_stores.helpers import ( _mmh3, @@ -42,11 +40,8 @@ compute_versioned_name, ) from feast.infra.online_stores.online_store import OnlineStore -from feast.infra.registry.base_registry import BaseRegistry from feast.infra.supported_async_methods import SupportedAsyncMethods -from feast.online_response import OnlineResponse from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto -from feast.protos.feast.types.Value_pb2 import RepeatedValue from feast.protos.feast.types.Value_pb2 import Value as ValueProto from feast.repo_config import FeastConfigBaseModel @@ -602,89 +597,27 @@ async def online_read_async( redis_values, fv_name, requested_features ) - def get_online_features( + def _read_features_per_fv( self, config: RepoConfig, - features: Union[List[str], FeatureService], - entity_rows: Union[ - List[Dict[str, Any]], - Mapping[str, Union[Sequence[Any], Sequence[ValueProto], RepeatedValue]], - ], - registry: BaseRegistry, - project: str, - full_feature_names: bool = False, - include_feature_view_version_metadata: bool = False, - ) -> OnlineResponse: - """ - Fetch online features for multiple feature views in a single Redis pipeline. - - Overrides the base class implementation which issues one pipeline per feature view - (O(N_feature_views) round trips). This implementation batches all HMGET commands - across all feature views into a single pipeline execution (O(1) round trips), - exploiting the fact that all feature views for the same entity share the same - Redis hash key. - """ - if isinstance(entity_rows, list): - columnar: Dict[str, List[Any]] = {k: [] for k in entity_rows[0].keys()} - for entity_row in entity_rows: - for key, value in entity_row.items(): - try: - columnar[key].append(value) - except KeyError as e: - raise ValueError( - "All entity_rows must have the same keys." - ) from e - entity_rows = columnar - - ( - join_key_values, - grouped_refs, - entity_name_to_join_key_map, - requested_on_demand_feature_views, - feature_refs, - requested_result_row_names, - online_features_response, - ) = utils._prepare_entities_to_read_from_online_store( - registry=registry, - project=project, - features=features, - entity_values=entity_rows, - full_feature_names=full_feature_names, - native_entity_values=True, - ) - - self._check_versioned_read_support(grouped_refs) - - _track_read = False - try: - from feast.metrics import _config as _metrics_config - - _track_read = _metrics_config.online_features - except Exception: - pass - - if _track_read: - import time as _time - - _read_start = _time.monotonic() - - # Pre-compute all Redis keys and hash field keys for every feature view so we - # can issue all HMGET commands in a single pipeline execution below. + grouped_refs, + join_key_values, + entity_name_to_join_key_map, + online_features_response, + full_feature_names: bool, + include_feature_view_version_metadata: bool, + ) -> None: + """Batch all per-FV HMGET commands into a single Redis pipeline.""" work_items = [] for table, requested_features in grouped_refs: table_entity_values, idxs, output_len = utils._get_unique_entities( - table, - join_key_values, - entity_name_to_join_key_map, + table, join_key_values, entity_name_to_join_key_map ) entity_key_protos = utils._get_entity_key_protos(table_entity_values) fv_name = _versioned_fv_name(table, config) redis_keys = self._generate_redis_keys_for_entities( config, entity_key_protos ) - - # Mutates requested_features in place (appends ts_key) — consistent with - # the base class behavior so _populate_response_from_feature_data works correctly. req_features, hset_keys = self._generate_hset_keys_for_features( table, requested_features, fv_name_override=fv_name ) @@ -692,7 +625,6 @@ def get_online_features( (table, req_features, fv_name, hset_keys, redis_keys, idxs, output_len) ) - # Single pipeline across all feature views: O(1) round trips instead of O(N_fv). if work_items: client = self._get_client(config.online_store) with client.pipeline(transaction=False) as pipe: @@ -730,96 +662,21 @@ def get_online_features( include_feature_view_version_metadata, ) - if _track_read: - from feast.metrics import track_online_store_read - - track_online_store_read(_time.monotonic() - _read_start) - - feature_types = self._build_feature_types(grouped_refs) - - if requested_on_demand_feature_views: - utils._augment_response_with_on_demand_transforms( - online_features_response, - feature_refs, - requested_on_demand_feature_views, - full_feature_names, - feature_types=feature_types, - ) - - utils._drop_unneeded_columns( - online_features_response, requested_result_row_names - ) - return OnlineResponse(online_features_response, feature_types=feature_types) - - async def get_online_features_async( + async def _read_features_per_fv_async( self, config: RepoConfig, - features: Union[List[str], FeatureService], - entity_rows: Union[ - List[Dict[str, Any]], - Mapping[str, Union[Sequence[Any], Sequence[ValueProto], RepeatedValue]], - ], - registry: BaseRegistry, - project: str, - full_feature_names: bool = False, - include_feature_view_version_metadata: bool = False, - ) -> OnlineResponse: - """ - Async version of get_online_features using a single batched Redis pipeline. - - Mirrors the sync override: all HMGET commands across all feature views are - issued in one async pipeline execution (O(1) round trips). - """ - if isinstance(entity_rows, list): - columnar: Dict[str, List[Any]] = {k: [] for k in entity_rows[0].keys()} - for entity_row in entity_rows: - for key, value in entity_row.items(): - try: - columnar[key].append(value) - except KeyError as e: - raise ValueError( - "All entity_rows must have the same keys." - ) from e - entity_rows = columnar - - ( - join_key_values, - grouped_refs, - entity_name_to_join_key_map, - requested_on_demand_feature_views, - feature_refs, - requested_result_row_names, - online_features_response, - ) = utils._prepare_entities_to_read_from_online_store( - registry=registry, - project=project, - features=features, - entity_values=entity_rows, - full_feature_names=full_feature_names, - native_entity_values=True, - ) - - self._check_versioned_read_support(grouped_refs) - - _track_read = False - try: - from feast.metrics import _config as _metrics_config - - _track_read = _metrics_config.online_features - except Exception: - pass - - if _track_read: - import time as _time - - _read_start = _time.monotonic() - + grouped_refs, + join_key_values, + entity_name_to_join_key_map, + online_features_response, + full_feature_names: bool, + include_feature_view_version_metadata: bool, + ) -> None: + """Async version: batch all per-FV HMGET into a single async pipeline.""" work_items = [] for table, requested_features in grouped_refs: table_entity_values, idxs, output_len = utils._get_unique_entities( - table, - join_key_values, - entity_name_to_join_key_map, + table, join_key_values, entity_name_to_join_key_map ) entity_key_protos = utils._get_entity_key_protos(table_entity_values) fv_name = _versioned_fv_name(table, config) @@ -870,27 +727,6 @@ async def get_online_features_async( include_feature_view_version_metadata, ) - if _track_read: - from feast.metrics import track_online_store_read - - track_online_store_read(_time.monotonic() - _read_start) - - feature_types = self._build_feature_types(grouped_refs) - - if requested_on_demand_feature_views: - utils._augment_response_with_on_demand_transforms( - online_features_response, - feature_refs, - requested_on_demand_feature_views, - full_feature_names, - feature_types=feature_types, - ) - - utils._drop_unneeded_columns( - online_features_response, requested_result_row_names - ) - return OnlineResponse(online_features_response, feature_types=feature_types) - def _get_features_for_entity( self, values: List[ByteString], diff --git a/sdk/python/feast/protos/feast/core/DataSource_pb2.pyi b/sdk/python/feast/protos/feast/core/DataSource_pb2.pyi index 3df12085cb6..36cd327a298 100644 --- a/sdk/python/feast/protos/feast/core/DataSource_pb2.pyi +++ b/sdk/python/feast/protos/feast/core/DataSource_pb2.pyi @@ -16,42 +16,40 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. """ - -from collections import abc as _abc -from feast.core import DataFormat_pb2 as _DataFormat_pb2 # type: ignore[attr-defined] -from feast.core import Feature_pb2 as _Feature_pb2 # type: ignore[attr-defined] -from feast.types import Value_pb2 as _Value_pb2 # type: ignore[attr-defined] -from google.protobuf import descriptor as _descriptor -from google.protobuf import duration_pb2 as _duration_pb2 -from google.protobuf import message as _message -from google.protobuf import timestamp_pb2 as _timestamp_pb2 -from google.protobuf.internal import containers as _containers -from google.protobuf.internal import enum_type_wrapper as _enum_type_wrapper -import builtins as _builtins +import builtins +import collections.abc +import feast.core.DataFormat_pb2 +import feast.core.Feature_pb2 +import feast.types.Value_pb2 +import google.protobuf.descriptor +import google.protobuf.duration_pb2 +import google.protobuf.internal.containers +import google.protobuf.internal.enum_type_wrapper +import google.protobuf.message +import google.protobuf.timestamp_pb2 import sys -import typing as _typing +import typing if sys.version_info >= (3, 10): - from typing import TypeAlias as _TypeAlias + import typing as typing_extensions else: - from typing_extensions import TypeAlias as _TypeAlias + import typing_extensions -DESCRIPTOR: _descriptor.FileDescriptor +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor -@_typing.final -class DataSource(_message.Message): +class DataSource(google.protobuf.message.Message): """Defines a Data Source that can be used source Feature data Next available id: 29 """ - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor class _SourceType: - ValueType = _typing.NewType("ValueType", _builtins.int) - V: _TypeAlias = ValueType # noqa: Y015 + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType - class _SourceTypeEnumTypeWrapper(_enum_type_wrapper._EnumTypeWrapper[DataSource._SourceType.ValueType], _builtins.type): - DESCRIPTOR: _descriptor.EnumDescriptor + class _SourceTypeEnumTypeWrapper(google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[DataSource._SourceType.ValueType], builtins.type): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor INVALID: DataSource._SourceType.ValueType # 0 BATCH_FILE: DataSource._SourceType.ValueType # 1 BATCH_SNOWFLAKE: DataSource._SourceType.ValueType # 8 @@ -85,565 +83,516 @@ class DataSource(_message.Message): BATCH_SPARK: DataSource.SourceType.ValueType # 11 BATCH_ATHENA: DataSource.SourceType.ValueType # 12 - @_typing.final - class TagsEntry(_message.Message): - DESCRIPTOR: _descriptor.Descriptor + class TagsEntry(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor - KEY_FIELD_NUMBER: _builtins.int - VALUE_FIELD_NUMBER: _builtins.int - key: _builtins.str - value: _builtins.str + KEY_FIELD_NUMBER: builtins.int + VALUE_FIELD_NUMBER: builtins.int + key: builtins.str + value: builtins.str def __init__( self, *, - key: _builtins.str = ..., - value: _builtins.str = ..., + key: builtins.str = ..., + value: builtins.str = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["key", b"key", "value", b"value"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["key", b"key", "value", b"value"]) -> None: ... - @_typing.final - class FieldMappingEntry(_message.Message): - DESCRIPTOR: _descriptor.Descriptor + class FieldMappingEntry(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor - KEY_FIELD_NUMBER: _builtins.int - VALUE_FIELD_NUMBER: _builtins.int - key: _builtins.str - value: _builtins.str + KEY_FIELD_NUMBER: builtins.int + VALUE_FIELD_NUMBER: builtins.int + key: builtins.str + value: builtins.str def __init__( self, *, - key: _builtins.str = ..., - value: _builtins.str = ..., + key: builtins.str = ..., + value: builtins.str = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["key", b"key", "value", b"value"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... - - @_typing.final - class SourceMeta(_message.Message): - DESCRIPTOR: _descriptor.Descriptor - - EARLIESTEVENTTIMESTAMP_FIELD_NUMBER: _builtins.int - LATESTEVENTTIMESTAMP_FIELD_NUMBER: _builtins.int - CREATED_TIMESTAMP_FIELD_NUMBER: _builtins.int - LAST_UPDATED_TIMESTAMP_FIELD_NUMBER: _builtins.int - @_builtins.property - def earliestEventTimestamp(self) -> _timestamp_pb2.Timestamp: ... - @_builtins.property - def latestEventTimestamp(self) -> _timestamp_pb2.Timestamp: ... - @_builtins.property - def created_timestamp(self) -> _timestamp_pb2.Timestamp: ... - @_builtins.property - def last_updated_timestamp(self) -> _timestamp_pb2.Timestamp: ... + def ClearField(self, field_name: typing_extensions.Literal["key", b"key", "value", b"value"]) -> None: ... + + class SourceMeta(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + EARLIESTEVENTTIMESTAMP_FIELD_NUMBER: builtins.int + LATESTEVENTTIMESTAMP_FIELD_NUMBER: builtins.int + CREATED_TIMESTAMP_FIELD_NUMBER: builtins.int + LAST_UPDATED_TIMESTAMP_FIELD_NUMBER: builtins.int + @property + def earliestEventTimestamp(self) -> google.protobuf.timestamp_pb2.Timestamp: ... + @property + def latestEventTimestamp(self) -> google.protobuf.timestamp_pb2.Timestamp: ... + @property + def created_timestamp(self) -> google.protobuf.timestamp_pb2.Timestamp: ... + @property + def last_updated_timestamp(self) -> google.protobuf.timestamp_pb2.Timestamp: ... def __init__( self, *, - earliestEventTimestamp: _timestamp_pb2.Timestamp | None = ..., - latestEventTimestamp: _timestamp_pb2.Timestamp | None = ..., - created_timestamp: _timestamp_pb2.Timestamp | None = ..., - last_updated_timestamp: _timestamp_pb2.Timestamp | None = ..., + earliestEventTimestamp: google.protobuf.timestamp_pb2.Timestamp | None = ..., + latestEventTimestamp: google.protobuf.timestamp_pb2.Timestamp | None = ..., + created_timestamp: google.protobuf.timestamp_pb2.Timestamp | None = ..., + last_updated_timestamp: google.protobuf.timestamp_pb2.Timestamp | None = ..., ) -> None: ... - _HasFieldArgType: _TypeAlias = _typing.Literal["created_timestamp", b"created_timestamp", "earliestEventTimestamp", b"earliestEventTimestamp", "last_updated_timestamp", b"last_updated_timestamp", "latestEventTimestamp", b"latestEventTimestamp"] # noqa: Y015 - def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["created_timestamp", b"created_timestamp", "earliestEventTimestamp", b"earliestEventTimestamp", "last_updated_timestamp", b"last_updated_timestamp", "latestEventTimestamp", b"latestEventTimestamp"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def HasField(self, field_name: typing_extensions.Literal["created_timestamp", b"created_timestamp", "earliestEventTimestamp", b"earliestEventTimestamp", "last_updated_timestamp", b"last_updated_timestamp", "latestEventTimestamp", b"latestEventTimestamp"]) -> builtins.bool: ... + def ClearField(self, field_name: typing_extensions.Literal["created_timestamp", b"created_timestamp", "earliestEventTimestamp", b"earliestEventTimestamp", "last_updated_timestamp", b"last_updated_timestamp", "latestEventTimestamp", b"latestEventTimestamp"]) -> None: ... - @_typing.final - class FileOptions(_message.Message): + class FileOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from a file""" - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - FILE_FORMAT_FIELD_NUMBER: _builtins.int - URI_FIELD_NUMBER: _builtins.int - S3_ENDPOINT_OVERRIDE_FIELD_NUMBER: _builtins.int - uri: _builtins.str + FILE_FORMAT_FIELD_NUMBER: builtins.int + URI_FIELD_NUMBER: builtins.int + S3_ENDPOINT_OVERRIDE_FIELD_NUMBER: builtins.int + @property + def file_format(self) -> feast.core.DataFormat_pb2.FileFormat: ... + uri: builtins.str """Target URL of file to retrieve and source features from. s3://path/to/file for AWS S3 storage gs://path/to/file for GCP GCS storage file:///path/to/file for local storage """ - s3_endpoint_override: _builtins.str + s3_endpoint_override: builtins.str """override AWS S3 storage endpoint with custom S3 endpoint""" - @_builtins.property - def file_format(self) -> _DataFormat_pb2.FileFormat: ... def __init__( self, *, - file_format: _DataFormat_pb2.FileFormat | None = ..., - uri: _builtins.str = ..., - s3_endpoint_override: _builtins.str = ..., + file_format: feast.core.DataFormat_pb2.FileFormat | None = ..., + uri: builtins.str = ..., + s3_endpoint_override: builtins.str = ..., ) -> None: ... - _HasFieldArgType: _TypeAlias = _typing.Literal["file_format", b"file_format"] # noqa: Y015 - def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["file_format", b"file_format", "s3_endpoint_override", b"s3_endpoint_override", "uri", b"uri"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def HasField(self, field_name: typing_extensions.Literal["file_format", b"file_format"]) -> builtins.bool: ... + def ClearField(self, field_name: typing_extensions.Literal["file_format", b"file_format", "s3_endpoint_override", b"s3_endpoint_override", "uri", b"uri"]) -> None: ... - @_typing.final - class BigQueryOptions(_message.Message): + class BigQueryOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from a BigQuery Query""" - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - TABLE_FIELD_NUMBER: _builtins.int - QUERY_FIELD_NUMBER: _builtins.int - table: _builtins.str + TABLE_FIELD_NUMBER: builtins.int + QUERY_FIELD_NUMBER: builtins.int + table: builtins.str """Full table reference in the form of [project:dataset.table]""" - query: _builtins.str + query: builtins.str """SQL query that returns a table containing feature data. Must contain an event_timestamp column, and respective entity columns """ def __init__( self, *, - table: _builtins.str = ..., - query: _builtins.str = ..., + table: builtins.str = ..., + query: builtins.str = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["query", b"query", "table", b"table"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["query", b"query", "table", b"table"]) -> None: ... - @_typing.final - class TrinoOptions(_message.Message): + class TrinoOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from a Trino Query""" - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - TABLE_FIELD_NUMBER: _builtins.int - QUERY_FIELD_NUMBER: _builtins.int - table: _builtins.str + TABLE_FIELD_NUMBER: builtins.int + QUERY_FIELD_NUMBER: builtins.int + table: builtins.str """Full table reference in the form of [project:dataset.table]""" - query: _builtins.str + query: builtins.str """SQL query that returns a table containing feature data. Must contain an event_timestamp column, and respective entity columns """ def __init__( self, *, - table: _builtins.str = ..., - query: _builtins.str = ..., + table: builtins.str = ..., + query: builtins.str = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["query", b"query", "table", b"table"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["query", b"query", "table", b"table"]) -> None: ... - @_typing.final - class KafkaOptions(_message.Message): + class KafkaOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from Kafka messages. Each message should be a Protobuf that can be decoded with the generated Java Protobuf class at the given class path """ - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - KAFKA_BOOTSTRAP_SERVERS_FIELD_NUMBER: _builtins.int - TOPIC_FIELD_NUMBER: _builtins.int - MESSAGE_FORMAT_FIELD_NUMBER: _builtins.int - WATERMARK_DELAY_THRESHOLD_FIELD_NUMBER: _builtins.int - kafka_bootstrap_servers: _builtins.str + KAFKA_BOOTSTRAP_SERVERS_FIELD_NUMBER: builtins.int + TOPIC_FIELD_NUMBER: builtins.int + MESSAGE_FORMAT_FIELD_NUMBER: builtins.int + WATERMARK_DELAY_THRESHOLD_FIELD_NUMBER: builtins.int + kafka_bootstrap_servers: builtins.str """Comma separated list of Kafka bootstrap servers. Used for feature tables without a defined source host[:port]]""" - topic: _builtins.str + topic: builtins.str """Kafka topic to collect feature data from.""" - @_builtins.property - def message_format(self) -> _DataFormat_pb2.StreamFormat: + @property + def message_format(self) -> feast.core.DataFormat_pb2.StreamFormat: """Defines the stream data format encoding feature/entity data in Kafka messages.""" - - @_builtins.property - def watermark_delay_threshold(self) -> _duration_pb2.Duration: + @property + def watermark_delay_threshold(self) -> google.protobuf.duration_pb2.Duration: """Watermark delay threshold for stream data""" - def __init__( self, *, - kafka_bootstrap_servers: _builtins.str = ..., - topic: _builtins.str = ..., - message_format: _DataFormat_pb2.StreamFormat | None = ..., - watermark_delay_threshold: _duration_pb2.Duration | None = ..., + kafka_bootstrap_servers: builtins.str = ..., + topic: builtins.str = ..., + message_format: feast.core.DataFormat_pb2.StreamFormat | None = ..., + watermark_delay_threshold: google.protobuf.duration_pb2.Duration | None = ..., ) -> None: ... - _HasFieldArgType: _TypeAlias = _typing.Literal["message_format", b"message_format", "watermark_delay_threshold", b"watermark_delay_threshold"] # noqa: Y015 - def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["kafka_bootstrap_servers", b"kafka_bootstrap_servers", "message_format", b"message_format", "topic", b"topic", "watermark_delay_threshold", b"watermark_delay_threshold"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def HasField(self, field_name: typing_extensions.Literal["message_format", b"message_format", "watermark_delay_threshold", b"watermark_delay_threshold"]) -> builtins.bool: ... + def ClearField(self, field_name: typing_extensions.Literal["kafka_bootstrap_servers", b"kafka_bootstrap_servers", "message_format", b"message_format", "topic", b"topic", "watermark_delay_threshold", b"watermark_delay_threshold"]) -> None: ... - @_typing.final - class KinesisOptions(_message.Message): + class KinesisOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from Kinesis records. Each record should be a Protobuf that can be decoded with the generated Java Protobuf class at the given class path """ - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - REGION_FIELD_NUMBER: _builtins.int - STREAM_NAME_FIELD_NUMBER: _builtins.int - RECORD_FORMAT_FIELD_NUMBER: _builtins.int - region: _builtins.str + REGION_FIELD_NUMBER: builtins.int + STREAM_NAME_FIELD_NUMBER: builtins.int + RECORD_FORMAT_FIELD_NUMBER: builtins.int + region: builtins.str """AWS region of the Kinesis stream""" - stream_name: _builtins.str + stream_name: builtins.str """Name of the Kinesis stream to obtain feature data from.""" - @_builtins.property - def record_format(self) -> _DataFormat_pb2.StreamFormat: + @property + def record_format(self) -> feast.core.DataFormat_pb2.StreamFormat: """Defines the data format encoding the feature/entity data in Kinesis records. Kinesis Data Sources support Avro and Proto as data formats. """ - def __init__( self, *, - region: _builtins.str = ..., - stream_name: _builtins.str = ..., - record_format: _DataFormat_pb2.StreamFormat | None = ..., + region: builtins.str = ..., + stream_name: builtins.str = ..., + record_format: feast.core.DataFormat_pb2.StreamFormat | None = ..., ) -> None: ... - _HasFieldArgType: _TypeAlias = _typing.Literal["record_format", b"record_format"] # noqa: Y015 - def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["record_format", b"record_format", "region", b"region", "stream_name", b"stream_name"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def HasField(self, field_name: typing_extensions.Literal["record_format", b"record_format"]) -> builtins.bool: ... + def ClearField(self, field_name: typing_extensions.Literal["record_format", b"record_format", "region", b"region", "stream_name", b"stream_name"]) -> None: ... - @_typing.final - class RedshiftOptions(_message.Message): + class RedshiftOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from a Redshift Query""" - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - TABLE_FIELD_NUMBER: _builtins.int - QUERY_FIELD_NUMBER: _builtins.int - SCHEMA_FIELD_NUMBER: _builtins.int - DATABASE_FIELD_NUMBER: _builtins.int - table: _builtins.str + TABLE_FIELD_NUMBER: builtins.int + QUERY_FIELD_NUMBER: builtins.int + SCHEMA_FIELD_NUMBER: builtins.int + DATABASE_FIELD_NUMBER: builtins.int + table: builtins.str """Redshift table name""" - query: _builtins.str + query: builtins.str """SQL query that returns a table containing feature data. Must contain an event_timestamp column, and respective entity columns """ - schema: _builtins.str + schema: builtins.str """Redshift schema name""" - database: _builtins.str + database: builtins.str """Redshift database name""" def __init__( self, *, - table: _builtins.str = ..., - query: _builtins.str = ..., - schema: _builtins.str = ..., - database: _builtins.str = ..., + table: builtins.str = ..., + query: builtins.str = ..., + schema: builtins.str = ..., + database: builtins.str = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["database", b"database", "query", b"query", "schema", b"schema", "table", b"table"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["database", b"database", "query", b"query", "schema", b"schema", "table", b"table"]) -> None: ... - @_typing.final - class AthenaOptions(_message.Message): + class AthenaOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from a Athena Query""" - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - TABLE_FIELD_NUMBER: _builtins.int - QUERY_FIELD_NUMBER: _builtins.int - DATABASE_FIELD_NUMBER: _builtins.int - DATA_SOURCE_FIELD_NUMBER: _builtins.int - table: _builtins.str + TABLE_FIELD_NUMBER: builtins.int + QUERY_FIELD_NUMBER: builtins.int + DATABASE_FIELD_NUMBER: builtins.int + DATA_SOURCE_FIELD_NUMBER: builtins.int + table: builtins.str """Athena table name""" - query: _builtins.str + query: builtins.str """SQL query that returns a table containing feature data. Must contain an event_timestamp column, and respective entity columns """ - database: _builtins.str + database: builtins.str """Athena database name""" - data_source: _builtins.str + data_source: builtins.str """Athena schema name""" def __init__( self, *, - table: _builtins.str = ..., - query: _builtins.str = ..., - database: _builtins.str = ..., - data_source: _builtins.str = ..., + table: builtins.str = ..., + query: builtins.str = ..., + database: builtins.str = ..., + data_source: builtins.str = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["data_source", b"data_source", "database", b"database", "query", b"query", "table", b"table"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["data_source", b"data_source", "database", b"database", "query", b"query", "table", b"table"]) -> None: ... - @_typing.final - class SnowflakeOptions(_message.Message): + class SnowflakeOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from a Snowflake Query""" - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - TABLE_FIELD_NUMBER: _builtins.int - QUERY_FIELD_NUMBER: _builtins.int - SCHEMA_FIELD_NUMBER: _builtins.int - DATABASE_FIELD_NUMBER: _builtins.int - table: _builtins.str + TABLE_FIELD_NUMBER: builtins.int + QUERY_FIELD_NUMBER: builtins.int + SCHEMA_FIELD_NUMBER: builtins.int + DATABASE_FIELD_NUMBER: builtins.int + table: builtins.str """Snowflake table name""" - query: _builtins.str + query: builtins.str """SQL query that returns a table containing feature data. Must contain an event_timestamp column, and respective entity columns """ - schema: _builtins.str + schema: builtins.str """Snowflake schema name""" - database: _builtins.str + database: builtins.str """Snowflake schema name""" def __init__( self, *, - table: _builtins.str = ..., - query: _builtins.str = ..., - schema: _builtins.str = ..., - database: _builtins.str = ..., + table: builtins.str = ..., + query: builtins.str = ..., + schema: builtins.str = ..., + database: builtins.str = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["database", b"database", "query", b"query", "schema", b"schema", "table", b"table"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["database", b"database", "query", b"query", "schema", b"schema", "table", b"table"]) -> None: ... - @_typing.final - class SparkOptions(_message.Message): + class SparkOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from a spark table/query""" - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - TABLE_FIELD_NUMBER: _builtins.int - QUERY_FIELD_NUMBER: _builtins.int - PATH_FIELD_NUMBER: _builtins.int - FILE_FORMAT_FIELD_NUMBER: _builtins.int - DATE_PARTITION_COLUMN_FORMAT_FIELD_NUMBER: _builtins.int - TABLE_FORMAT_FIELD_NUMBER: _builtins.int - table: _builtins.str + TABLE_FIELD_NUMBER: builtins.int + QUERY_FIELD_NUMBER: builtins.int + PATH_FIELD_NUMBER: builtins.int + FILE_FORMAT_FIELD_NUMBER: builtins.int + DATE_PARTITION_COLUMN_FORMAT_FIELD_NUMBER: builtins.int + TABLE_FORMAT_FIELD_NUMBER: builtins.int + table: builtins.str """Table name""" - query: _builtins.str + query: builtins.str """Spark SQl query that returns the table, this is an alternative to `table`""" - path: _builtins.str + path: builtins.str """Path from which spark can read the table, this is an alternative to `table`""" - file_format: _builtins.str + file_format: builtins.str """Format of files at `path` (e.g. parquet, avro, etc)""" - date_partition_column_format: _builtins.str + date_partition_column_format: builtins.str """Date Format of date partition column (e.g. %Y-%m-%d)""" - @_builtins.property - def table_format(self) -> _DataFormat_pb2.TableFormat: + @property + def table_format(self) -> feast.core.DataFormat_pb2.TableFormat: """Table Format (e.g. iceberg, delta, hudi)""" - def __init__( self, *, - table: _builtins.str = ..., - query: _builtins.str = ..., - path: _builtins.str = ..., - file_format: _builtins.str = ..., - date_partition_column_format: _builtins.str = ..., - table_format: _DataFormat_pb2.TableFormat | None = ..., + table: builtins.str = ..., + query: builtins.str = ..., + path: builtins.str = ..., + file_format: builtins.str = ..., + date_partition_column_format: builtins.str = ..., + table_format: feast.core.DataFormat_pb2.TableFormat | None = ..., ) -> None: ... - _HasFieldArgType: _TypeAlias = _typing.Literal["table_format", b"table_format"] # noqa: Y015 - def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["date_partition_column_format", b"date_partition_column_format", "file_format", b"file_format", "path", b"path", "query", b"query", "table", b"table", "table_format", b"table_format"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def HasField(self, field_name: typing_extensions.Literal["table_format", b"table_format"]) -> builtins.bool: ... + def ClearField(self, field_name: typing_extensions.Literal["date_partition_column_format", b"date_partition_column_format", "file_format", b"file_format", "path", b"path", "query", b"query", "table", b"table", "table_format", b"table_format"]) -> None: ... - @_typing.final - class CustomSourceOptions(_message.Message): + class CustomSourceOptions(google.protobuf.message.Message): """Defines configuration for custom third-party data sources.""" - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - CONFIGURATION_FIELD_NUMBER: _builtins.int - configuration: _builtins.bytes + CONFIGURATION_FIELD_NUMBER: builtins.int + configuration: builtins.bytes """Serialized configuration information for the data source. The implementer of the custom data source is responsible for serializing and deserializing data from bytes """ def __init__( self, *, - configuration: _builtins.bytes = ..., + configuration: builtins.bytes = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["configuration", b"configuration"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["configuration", b"configuration"]) -> None: ... - @_typing.final - class RequestDataOptions(_message.Message): + class RequestDataOptions(google.protobuf.message.Message): """Defines options for DataSource that sources features from request data""" - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor - @_typing.final - class DeprecatedSchemaEntry(_message.Message): - DESCRIPTOR: _descriptor.Descriptor + class DeprecatedSchemaEntry(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor - KEY_FIELD_NUMBER: _builtins.int - VALUE_FIELD_NUMBER: _builtins.int - key: _builtins.str - value: _Value_pb2.ValueType.Enum.ValueType + KEY_FIELD_NUMBER: builtins.int + VALUE_FIELD_NUMBER: builtins.int + key: builtins.str + value: feast.types.Value_pb2.ValueType.Enum.ValueType def __init__( self, *, - key: _builtins.str = ..., - value: _Value_pb2.ValueType.Enum.ValueType = ..., + key: builtins.str = ..., + value: feast.types.Value_pb2.ValueType.Enum.ValueType = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["key", b"key", "value", b"value"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["key", b"key", "value", b"value"]) -> None: ... - DEPRECATED_SCHEMA_FIELD_NUMBER: _builtins.int - SCHEMA_FIELD_NUMBER: _builtins.int - @_builtins.property - def deprecated_schema(self) -> _containers.ScalarMap[_builtins.str, _Value_pb2.ValueType.Enum.ValueType]: + DEPRECATED_SCHEMA_FIELD_NUMBER: builtins.int + SCHEMA_FIELD_NUMBER: builtins.int + @property + def deprecated_schema(self) -> google.protobuf.internal.containers.ScalarMap[builtins.str, feast.types.Value_pb2.ValueType.Enum.ValueType]: """Mapping of feature name to type""" - - @_builtins.property - def schema(self) -> _containers.RepeatedCompositeFieldContainer[_Feature_pb2.FeatureSpecV2]: ... + @property + def schema(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[feast.core.Feature_pb2.FeatureSpecV2]: ... def __init__( self, *, - deprecated_schema: _abc.Mapping[_builtins.str, _Value_pb2.ValueType.Enum.ValueType] | None = ..., - schema: _abc.Iterable[_Feature_pb2.FeatureSpecV2] | None = ..., + deprecated_schema: collections.abc.Mapping[builtins.str, feast.types.Value_pb2.ValueType.Enum.ValueType] | None = ..., + schema: collections.abc.Iterable[feast.core.Feature_pb2.FeatureSpecV2] | None = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["deprecated_schema", b"deprecated_schema", "schema", b"schema"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["deprecated_schema", b"deprecated_schema", "schema", b"schema"]) -> None: ... - @_typing.final - class PushOptions(_message.Message): + class PushOptions(google.protobuf.message.Message): """Defines options for DataSource that supports pushing data to it. This allows data to be pushed to the online store on-demand, such as by stream consumers. """ - DESCRIPTOR: _descriptor.Descriptor + DESCRIPTOR: google.protobuf.descriptor.Descriptor def __init__( self, ) -> None: ... - NAME_FIELD_NUMBER: _builtins.int - PROJECT_FIELD_NUMBER: _builtins.int - DESCRIPTION_FIELD_NUMBER: _builtins.int - TAGS_FIELD_NUMBER: _builtins.int - OWNER_FIELD_NUMBER: _builtins.int - TYPE_FIELD_NUMBER: _builtins.int - FIELD_MAPPING_FIELD_NUMBER: _builtins.int - TIMESTAMP_FIELD_FIELD_NUMBER: _builtins.int - DATE_PARTITION_COLUMN_FIELD_NUMBER: _builtins.int - CREATED_TIMESTAMP_COLUMN_FIELD_NUMBER: _builtins.int - DATA_SOURCE_CLASS_TYPE_FIELD_NUMBER: _builtins.int - BATCH_SOURCE_FIELD_NUMBER: _builtins.int - META_FIELD_NUMBER: _builtins.int - FILE_OPTIONS_FIELD_NUMBER: _builtins.int - BIGQUERY_OPTIONS_FIELD_NUMBER: _builtins.int - KAFKA_OPTIONS_FIELD_NUMBER: _builtins.int - KINESIS_OPTIONS_FIELD_NUMBER: _builtins.int - REDSHIFT_OPTIONS_FIELD_NUMBER: _builtins.int - REQUEST_DATA_OPTIONS_FIELD_NUMBER: _builtins.int - CUSTOM_OPTIONS_FIELD_NUMBER: _builtins.int - SNOWFLAKE_OPTIONS_FIELD_NUMBER: _builtins.int - PUSH_OPTIONS_FIELD_NUMBER: _builtins.int - SPARK_OPTIONS_FIELD_NUMBER: _builtins.int - TRINO_OPTIONS_FIELD_NUMBER: _builtins.int - ATHENA_OPTIONS_FIELD_NUMBER: _builtins.int - TIMESTAMP_FIELD_TYPE_FIELD_NUMBER: _builtins.int - name: _builtins.str + NAME_FIELD_NUMBER: builtins.int + PROJECT_FIELD_NUMBER: builtins.int + DESCRIPTION_FIELD_NUMBER: builtins.int + TAGS_FIELD_NUMBER: builtins.int + OWNER_FIELD_NUMBER: builtins.int + TYPE_FIELD_NUMBER: builtins.int + FIELD_MAPPING_FIELD_NUMBER: builtins.int + TIMESTAMP_FIELD_FIELD_NUMBER: builtins.int + DATE_PARTITION_COLUMN_FIELD_NUMBER: builtins.int + CREATED_TIMESTAMP_COLUMN_FIELD_NUMBER: builtins.int + TIMESTAMP_FIELD_TYPE_FIELD_NUMBER: builtins.int + DATA_SOURCE_CLASS_TYPE_FIELD_NUMBER: builtins.int + BATCH_SOURCE_FIELD_NUMBER: builtins.int + META_FIELD_NUMBER: builtins.int + FILE_OPTIONS_FIELD_NUMBER: builtins.int + BIGQUERY_OPTIONS_FIELD_NUMBER: builtins.int + KAFKA_OPTIONS_FIELD_NUMBER: builtins.int + KINESIS_OPTIONS_FIELD_NUMBER: builtins.int + REDSHIFT_OPTIONS_FIELD_NUMBER: builtins.int + REQUEST_DATA_OPTIONS_FIELD_NUMBER: builtins.int + CUSTOM_OPTIONS_FIELD_NUMBER: builtins.int + SNOWFLAKE_OPTIONS_FIELD_NUMBER: builtins.int + PUSH_OPTIONS_FIELD_NUMBER: builtins.int + SPARK_OPTIONS_FIELD_NUMBER: builtins.int + TRINO_OPTIONS_FIELD_NUMBER: builtins.int + ATHENA_OPTIONS_FIELD_NUMBER: builtins.int + name: builtins.str """Unique name of data source within the project""" - project: _builtins.str + project: builtins.str """Name of Feast project that this data source belongs to.""" - description: _builtins.str - owner: _builtins.str - type: Global___DataSource.SourceType.ValueType - timestamp_field: _builtins.str + description: builtins.str + @property + def tags(self) -> google.protobuf.internal.containers.ScalarMap[builtins.str, builtins.str]: ... + owner: builtins.str + type: global___DataSource.SourceType.ValueType + @property + def field_mapping(self) -> google.protobuf.internal.containers.ScalarMap[builtins.str, builtins.str]: + """Defines mapping between fields in the sourced data + and fields in parent FeatureTable. + """ + timestamp_field: builtins.str """Must specify event timestamp column name""" - date_partition_column: _builtins.str + date_partition_column: builtins.str """(Optional) Specify partition column useful for file sources """ - created_timestamp_column: _builtins.str + created_timestamp_column: builtins.str """Must specify creation timestamp column name""" - timestamp_field_type: _builtins.str + timestamp_field_type: builtins.str """(Optional) Type of the timestamp_field column ("TIMESTAMP" or "DATE"). When set to "DATE", SQL generation uses date-only comparisons. """ - data_source_class_type: _builtins.str + data_source_class_type: builtins.str """This is an internal field that is represents the python class for the data source object a proto object represents. This should be set by feast, and not by users. The field is used primarily by custom data sources and is mandatory for them to set. Feast may set it for first party sources as well. """ - @_builtins.property - def tags(self) -> _containers.ScalarMap[_builtins.str, _builtins.str]: ... - @_builtins.property - def field_mapping(self) -> _containers.ScalarMap[_builtins.str, _builtins.str]: - """Defines mapping between fields in the sourced data - and fields in parent FeatureTable. - """ - - @_builtins.property - def batch_source(self) -> Global___DataSource: + @property + def batch_source(self) -> global___DataSource: """Optional batch source for streaming sources for historical features and materialization.""" - - @_builtins.property - def meta(self) -> Global___DataSource.SourceMeta: ... - @_builtins.property - def file_options(self) -> Global___DataSource.FileOptions: ... - @_builtins.property - def bigquery_options(self) -> Global___DataSource.BigQueryOptions: ... - @_builtins.property - def kafka_options(self) -> Global___DataSource.KafkaOptions: ... - @_builtins.property - def kinesis_options(self) -> Global___DataSource.KinesisOptions: ... - @_builtins.property - def redshift_options(self) -> Global___DataSource.RedshiftOptions: ... - @_builtins.property - def request_data_options(self) -> Global___DataSource.RequestDataOptions: ... - @_builtins.property - def custom_options(self) -> Global___DataSource.CustomSourceOptions: ... - @_builtins.property - def snowflake_options(self) -> Global___DataSource.SnowflakeOptions: ... - @_builtins.property - def push_options(self) -> Global___DataSource.PushOptions: ... - @_builtins.property - def spark_options(self) -> Global___DataSource.SparkOptions: ... - @_builtins.property - def trino_options(self) -> Global___DataSource.TrinoOptions: ... - @_builtins.property - def athena_options(self) -> Global___DataSource.AthenaOptions: ... + @property + def meta(self) -> global___DataSource.SourceMeta: ... + @property + def file_options(self) -> global___DataSource.FileOptions: ... + @property + def bigquery_options(self) -> global___DataSource.BigQueryOptions: ... + @property + def kafka_options(self) -> global___DataSource.KafkaOptions: ... + @property + def kinesis_options(self) -> global___DataSource.KinesisOptions: ... + @property + def redshift_options(self) -> global___DataSource.RedshiftOptions: ... + @property + def request_data_options(self) -> global___DataSource.RequestDataOptions: ... + @property + def custom_options(self) -> global___DataSource.CustomSourceOptions: ... + @property + def snowflake_options(self) -> global___DataSource.SnowflakeOptions: ... + @property + def push_options(self) -> global___DataSource.PushOptions: ... + @property + def spark_options(self) -> global___DataSource.SparkOptions: ... + @property + def trino_options(self) -> global___DataSource.TrinoOptions: ... + @property + def athena_options(self) -> global___DataSource.AthenaOptions: ... def __init__( self, *, - name: _builtins.str = ..., - project: _builtins.str = ..., - description: _builtins.str = ..., - tags: _abc.Mapping[_builtins.str, _builtins.str] | None = ..., - owner: _builtins.str = ..., - type: Global___DataSource.SourceType.ValueType = ..., - field_mapping: _abc.Mapping[_builtins.str, _builtins.str] | None = ..., - timestamp_field: _builtins.str = ..., - date_partition_column: _builtins.str = ..., - created_timestamp_column: _builtins.str = ..., - timestamp_field_type: _builtins.str = ..., - data_source_class_type: _builtins.str = ..., - batch_source: Global___DataSource | None = ..., - meta: Global___DataSource.SourceMeta | None = ..., - file_options: Global___DataSource.FileOptions | None = ..., - bigquery_options: Global___DataSource.BigQueryOptions | None = ..., - kafka_options: Global___DataSource.KafkaOptions | None = ..., - kinesis_options: Global___DataSource.KinesisOptions | None = ..., - redshift_options: Global___DataSource.RedshiftOptions | None = ..., - request_data_options: Global___DataSource.RequestDataOptions | None = ..., - custom_options: Global___DataSource.CustomSourceOptions | None = ..., - snowflake_options: Global___DataSource.SnowflakeOptions | None = ..., - push_options: Global___DataSource.PushOptions | None = ..., - spark_options: Global___DataSource.SparkOptions | None = ..., - trino_options: Global___DataSource.TrinoOptions | None = ..., - athena_options: Global___DataSource.AthenaOptions | None = ..., + name: builtins.str = ..., + project: builtins.str = ..., + description: builtins.str = ..., + tags: collections.abc.Mapping[builtins.str, builtins.str] | None = ..., + owner: builtins.str = ..., + type: global___DataSource.SourceType.ValueType = ..., + field_mapping: collections.abc.Mapping[builtins.str, builtins.str] | None = ..., + timestamp_field: builtins.str = ..., + date_partition_column: builtins.str = ..., + created_timestamp_column: builtins.str = ..., + timestamp_field_type: builtins.str = ..., + data_source_class_type: builtins.str = ..., + batch_source: global___DataSource | None = ..., + meta: global___DataSource.SourceMeta | None = ..., + file_options: global___DataSource.FileOptions | None = ..., + bigquery_options: global___DataSource.BigQueryOptions | None = ..., + kafka_options: global___DataSource.KafkaOptions | None = ..., + kinesis_options: global___DataSource.KinesisOptions | None = ..., + redshift_options: global___DataSource.RedshiftOptions | None = ..., + request_data_options: global___DataSource.RequestDataOptions | None = ..., + custom_options: global___DataSource.CustomSourceOptions | None = ..., + snowflake_options: global___DataSource.SnowflakeOptions | None = ..., + push_options: global___DataSource.PushOptions | None = ..., + spark_options: global___DataSource.SparkOptions | None = ..., + trino_options: global___DataSource.TrinoOptions | None = ..., + athena_options: global___DataSource.AthenaOptions | None = ..., ) -> None: ... - _HasFieldArgType: _TypeAlias = _typing.Literal["athena_options", b"athena_options", "batch_source", b"batch_source", "bigquery_options", b"bigquery_options", "custom_options", b"custom_options", "file_options", b"file_options", "kafka_options", b"kafka_options", "kinesis_options", b"kinesis_options", "meta", b"meta", "options", b"options", "push_options", b"push_options", "redshift_options", b"redshift_options", "request_data_options", b"request_data_options", "snowflake_options", b"snowflake_options", "spark_options", b"spark_options", "trino_options", b"trino_options"] # noqa: Y015 - def HasField(self, field_name: _HasFieldArgType) -> _builtins.bool: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["athena_options", b"athena_options", "batch_source", b"batch_source", "bigquery_options", b"bigquery_options", "created_timestamp_column", b"created_timestamp_column", "custom_options", b"custom_options", "data_source_class_type", b"data_source_class_type", "date_partition_column", b"date_partition_column", "description", b"description", "field_mapping", b"field_mapping", "file_options", b"file_options", "kafka_options", b"kafka_options", "kinesis_options", b"kinesis_options", "meta", b"meta", "name", b"name", "options", b"options", "owner", b"owner", "project", b"project", "push_options", b"push_options", "redshift_options", b"redshift_options", "request_data_options", b"request_data_options", "snowflake_options", b"snowflake_options", "spark_options", b"spark_options", "tags", b"tags", "timestamp_field", b"timestamp_field", "timestamp_field_type", b"timestamp_field_type", "trino_options", b"trino_options", "type", b"type"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... - _WhichOneofReturnType_options: _TypeAlias = _typing.Literal["file_options", "bigquery_options", "kafka_options", "kinesis_options", "redshift_options", "request_data_options", "custom_options", "snowflake_options", "push_options", "spark_options", "trino_options", "athena_options"] # noqa: Y015 - _WhichOneofArgType_options: _TypeAlias = _typing.Literal["options", b"options"] # noqa: Y015 - def WhichOneof(self, oneof_group: _WhichOneofArgType_options) -> _WhichOneofReturnType_options | None: ... - -Global___DataSource: _TypeAlias = DataSource # noqa: Y015 - -@_typing.final -class DataSourceList(_message.Message): - DESCRIPTOR: _descriptor.Descriptor - - DATASOURCES_FIELD_NUMBER: _builtins.int - @_builtins.property - def datasources(self) -> _containers.RepeatedCompositeFieldContainer[Global___DataSource]: ... + def HasField(self, field_name: typing_extensions.Literal["athena_options", b"athena_options", "batch_source", b"batch_source", "bigquery_options", b"bigquery_options", "custom_options", b"custom_options", "file_options", b"file_options", "kafka_options", b"kafka_options", "kinesis_options", b"kinesis_options", "meta", b"meta", "options", b"options", "push_options", b"push_options", "redshift_options", b"redshift_options", "request_data_options", b"request_data_options", "snowflake_options", b"snowflake_options", "spark_options", b"spark_options", "trino_options", b"trino_options"]) -> builtins.bool: ... + def ClearField(self, field_name: typing_extensions.Literal["athena_options", b"athena_options", "batch_source", b"batch_source", "bigquery_options", b"bigquery_options", "created_timestamp_column", b"created_timestamp_column", "custom_options", b"custom_options", "data_source_class_type", b"data_source_class_type", "date_partition_column", b"date_partition_column", "description", b"description", "field_mapping", b"field_mapping", "file_options", b"file_options", "kafka_options", b"kafka_options", "kinesis_options", b"kinesis_options", "meta", b"meta", "name", b"name", "options", b"options", "owner", b"owner", "project", b"project", "push_options", b"push_options", "redshift_options", b"redshift_options", "request_data_options", b"request_data_options", "snowflake_options", b"snowflake_options", "spark_options", b"spark_options", "tags", b"tags", "timestamp_field", b"timestamp_field", "timestamp_field_type", b"timestamp_field_type", "trino_options", b"trino_options", "type", b"type"]) -> None: ... + def WhichOneof(self, oneof_group: typing_extensions.Literal["options", b"options"]) -> typing_extensions.Literal["file_options", "bigquery_options", "kafka_options", "kinesis_options", "redshift_options", "request_data_options", "custom_options", "snowflake_options", "push_options", "spark_options", "trino_options", "athena_options"] | None: ... + +global___DataSource = DataSource + +class DataSourceList(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + DATASOURCES_FIELD_NUMBER: builtins.int + @property + def datasources(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___DataSource]: ... def __init__( self, *, - datasources: _abc.Iterable[Global___DataSource] | None = ..., + datasources: collections.abc.Iterable[global___DataSource] | None = ..., ) -> None: ... - _ClearFieldArgType: _TypeAlias = _typing.Literal["datasources", b"datasources"] # noqa: Y015 - def ClearField(self, field_name: _ClearFieldArgType) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["datasources", b"datasources"]) -> None: ... -Global___DataSourceList: _TypeAlias = DataSourceList # noqa: Y015 +global___DataSourceList = DataSourceList diff --git a/sdk/python/feast/protos/feast/core/FeatureService_pb2.py b/sdk/python/feast/protos/feast/core/FeatureService_pb2.py index 7ef36079691..24940b4ca33 100644 --- a/sdk/python/feast/protos/feast/core/FeatureService_pb2.py +++ b/sdk/python/feast/protos/feast/core/FeatureService_pb2.py @@ -16,7 +16,7 @@ from feast.protos.feast.core import FeatureViewProjection_pb2 as feast_dot_core_dot_FeatureViewProjection__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1f\x66\x65\x61st/core/FeatureService.proto\x12\nfeast.core\x1a\x1fgoogle/protobuf/timestamp.proto\x1a&feast/core/FeatureViewProjection.proto\"l\n\x0e\x46\x65\x61tureService\x12,\n\x04spec\x18\x01 \x01(\x0b\x32\x1e.feast.core.FeatureServiceSpec\x12,\n\x04meta\x18\x02 \x01(\x0b\x32\x1e.feast.core.FeatureServiceMeta\"\xa4\x02\n\x12\x46\x65\x61tureServiceSpec\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x0f\n\x07project\x18\x02 \x01(\t\x12\x33\n\x08\x66\x65\x61tures\x18\x03 \x03(\x0b\x32!.feast.core.FeatureViewProjection\x12\x36\n\x04tags\x18\x04 \x03(\x0b\x32(.feast.core.FeatureServiceSpec.TagsEntry\x12\x13\n\x0b\x64\x65scription\x18\x05 \x01(\t\x12\r\n\x05owner\x18\x06 \x01(\t\x12\x31\n\x0elogging_config\x18\x07 \x01(\x0b\x32\x19.feast.core.LoggingConfig\x1a+\n\tTagsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"\x87\x01\n\x12\x46\x65\x61tureServiceMeta\x12\x35\n\x11\x63reated_timestamp\x18\x01 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12:\n\x16last_updated_timestamp\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\"\xd1\x08\n\rLoggingConfig\x12\x13\n\x0bsample_rate\x18\x01 \x01(\x02\x12\x45\n\x10\x66ile_destination\x18\x03 \x01(\x0b\x32).feast.core.LoggingConfig.FileDestinationH\x00\x12M\n\x14\x62igquery_destination\x18\x04 \x01(\x0b\x32-.feast.core.LoggingConfig.BigQueryDestinationH\x00\x12M\n\x14redshift_destination\x18\x05 \x01(\x0b\x32-.feast.core.LoggingConfig.RedshiftDestinationH\x00\x12O\n\x15snowflake_destination\x18\x06 \x01(\x0b\x32..feast.core.LoggingConfig.SnowflakeDestinationH\x00\x12I\n\x12\x63ustom_destination\x18\x07 \x01(\x0b\x32+.feast.core.LoggingConfig.CustomDestinationH\x00\x12I\n\x12\x61thena_destination\x18\x08 \x01(\x0b\x32+.feast.core.LoggingConfig.AthenaDestinationH\x00\x12`\n\x1e\x63ouchbase_columnar_destination\x18\t \x01(\x0b\x32\x36.feast.core.LoggingConfig.CouchbaseColumnarDestinationH\x00\x1aS\n\x0f\x46ileDestination\x12\x0c\n\x04path\x18\x01 \x01(\t\x12\x1c\n\x14s3_endpoint_override\x18\x02 \x01(\t\x12\x14\n\x0cpartition_by\x18\x03 \x03(\t\x1a(\n\x13\x42igQueryDestination\x12\x11\n\ttable_ref\x18\x01 \x01(\t\x1a)\n\x13RedshiftDestination\x12\x12\n\ntable_name\x18\x01 \x01(\t\x1a\'\n\x11\x41thenaDestination\x12\x12\n\ntable_name\x18\x01 \x01(\t\x1a*\n\x14SnowflakeDestination\x12\x12\n\ntable_name\x18\x01 \x01(\t\x1a\x99\x01\n\x11\x43ustomDestination\x12\x0c\n\x04kind\x18\x01 \x01(\t\x12G\n\x06\x63onfig\x18\x02 \x03(\x0b\x32\x37.feast.core.LoggingConfig.CustomDestination.ConfigEntry\x1a-\n\x0b\x43onfigEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1aS\n\x1c\x43ouchbaseColumnarDestination\x12\x10\n\x08\x64\x61tabase\x18\x01 \x01(\t\x12\r\n\x05scope\x18\x02 \x01(\t\x12\x12\n\ncollection\x18\x03 \x01(\tB\r\n\x0b\x64\x65stination\"I\n\x12\x46\x65\x61tureServiceList\x12\x33\n\x0f\x66\x65\x61tureservices\x18\x01 \x03(\x0b\x32\x1a.feast.core.FeatureServiceBX\n\x10\x66\x65\x61st.proto.coreB\x13\x46\x65\x61tureServiceProtoZ/github.com/feast-dev/feast/go/protos/feast/coreb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1f\x66\x65\x61st/core/FeatureService.proto\x12\nfeast.core\x1a\x1fgoogle/protobuf/timestamp.proto\x1a&feast/core/FeatureViewProjection.proto\"l\n\x0e\x46\x65\x61tureService\x12,\n\x04spec\x18\x01 \x01(\x0b\x32\x1e.feast.core.FeatureServiceSpec\x12,\n\x04meta\x18\x02 \x01(\x0b\x32\x1e.feast.core.FeatureServiceMeta\"\xbf\x02\n\x12\x46\x65\x61tureServiceSpec\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x0f\n\x07project\x18\x02 \x01(\t\x12\x33\n\x08\x66\x65\x61tures\x18\x03 \x03(\x0b\x32!.feast.core.FeatureViewProjection\x12\x36\n\x04tags\x18\x04 \x03(\x0b\x32(.feast.core.FeatureServiceSpec.TagsEntry\x12\x13\n\x0b\x64\x65scription\x18\x05 \x01(\t\x12\r\n\x05owner\x18\x06 \x01(\t\x12\x31\n\x0elogging_config\x18\x07 \x01(\x0b\x32\x19.feast.core.LoggingConfig\x12\x19\n\x11precompute_online\x18\x08 \x01(\x08\x1a+\n\tTagsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"\x87\x01\n\x12\x46\x65\x61tureServiceMeta\x12\x35\n\x11\x63reated_timestamp\x18\x01 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12:\n\x16last_updated_timestamp\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\"\xd1\x08\n\rLoggingConfig\x12\x13\n\x0bsample_rate\x18\x01 \x01(\x02\x12\x45\n\x10\x66ile_destination\x18\x03 \x01(\x0b\x32).feast.core.LoggingConfig.FileDestinationH\x00\x12M\n\x14\x62igquery_destination\x18\x04 \x01(\x0b\x32-.feast.core.LoggingConfig.BigQueryDestinationH\x00\x12M\n\x14redshift_destination\x18\x05 \x01(\x0b\x32-.feast.core.LoggingConfig.RedshiftDestinationH\x00\x12O\n\x15snowflake_destination\x18\x06 \x01(\x0b\x32..feast.core.LoggingConfig.SnowflakeDestinationH\x00\x12I\n\x12\x63ustom_destination\x18\x07 \x01(\x0b\x32+.feast.core.LoggingConfig.CustomDestinationH\x00\x12I\n\x12\x61thena_destination\x18\x08 \x01(\x0b\x32+.feast.core.LoggingConfig.AthenaDestinationH\x00\x12`\n\x1e\x63ouchbase_columnar_destination\x18\t \x01(\x0b\x32\x36.feast.core.LoggingConfig.CouchbaseColumnarDestinationH\x00\x1aS\n\x0f\x46ileDestination\x12\x0c\n\x04path\x18\x01 \x01(\t\x12\x1c\n\x14s3_endpoint_override\x18\x02 \x01(\t\x12\x14\n\x0cpartition_by\x18\x03 \x03(\t\x1a(\n\x13\x42igQueryDestination\x12\x11\n\ttable_ref\x18\x01 \x01(\t\x1a)\n\x13RedshiftDestination\x12\x12\n\ntable_name\x18\x01 \x01(\t\x1a\'\n\x11\x41thenaDestination\x12\x12\n\ntable_name\x18\x01 \x01(\t\x1a*\n\x14SnowflakeDestination\x12\x12\n\ntable_name\x18\x01 \x01(\t\x1a\x99\x01\n\x11\x43ustomDestination\x12\x0c\n\x04kind\x18\x01 \x01(\t\x12G\n\x06\x63onfig\x18\x02 \x03(\x0b\x32\x37.feast.core.LoggingConfig.CustomDestination.ConfigEntry\x1a-\n\x0b\x43onfigEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1aS\n\x1c\x43ouchbaseColumnarDestination\x12\x10\n\x08\x64\x61tabase\x18\x01 \x01(\t\x12\r\n\x05scope\x18\x02 \x01(\t\x12\x12\n\ncollection\x18\x03 \x01(\tB\r\n\x0b\x64\x65stination\"I\n\x12\x46\x65\x61tureServiceList\x12\x33\n\x0f\x66\x65\x61tureservices\x18\x01 \x03(\x0b\x32\x1a.feast.core.FeatureServiceBX\n\x10\x66\x65\x61st.proto.coreB\x13\x46\x65\x61tureServiceProtoZ/github.com/feast-dev/feast/go/protos/feast/coreb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -31,29 +31,29 @@ _globals['_FEATURESERVICE']._serialized_start=120 _globals['_FEATURESERVICE']._serialized_end=228 _globals['_FEATURESERVICESPEC']._serialized_start=231 - _globals['_FEATURESERVICESPEC']._serialized_end=523 - _globals['_FEATURESERVICESPEC_TAGSENTRY']._serialized_start=480 - _globals['_FEATURESERVICESPEC_TAGSENTRY']._serialized_end=523 - _globals['_FEATURESERVICEMETA']._serialized_start=526 - _globals['_FEATURESERVICEMETA']._serialized_end=661 - _globals['_LOGGINGCONFIG']._serialized_start=664 - _globals['_LOGGINGCONFIG']._serialized_end=1769 - _globals['_LOGGINGCONFIG_FILEDESTINATION']._serialized_start=1260 - _globals['_LOGGINGCONFIG_FILEDESTINATION']._serialized_end=1343 - _globals['_LOGGINGCONFIG_BIGQUERYDESTINATION']._serialized_start=1345 - _globals['_LOGGINGCONFIG_BIGQUERYDESTINATION']._serialized_end=1385 - _globals['_LOGGINGCONFIG_REDSHIFTDESTINATION']._serialized_start=1387 - _globals['_LOGGINGCONFIG_REDSHIFTDESTINATION']._serialized_end=1428 - _globals['_LOGGINGCONFIG_ATHENADESTINATION']._serialized_start=1430 - _globals['_LOGGINGCONFIG_ATHENADESTINATION']._serialized_end=1469 - _globals['_LOGGINGCONFIG_SNOWFLAKEDESTINATION']._serialized_start=1471 - _globals['_LOGGINGCONFIG_SNOWFLAKEDESTINATION']._serialized_end=1513 - _globals['_LOGGINGCONFIG_CUSTOMDESTINATION']._serialized_start=1516 - _globals['_LOGGINGCONFIG_CUSTOMDESTINATION']._serialized_end=1669 - _globals['_LOGGINGCONFIG_CUSTOMDESTINATION_CONFIGENTRY']._serialized_start=1624 - _globals['_LOGGINGCONFIG_CUSTOMDESTINATION_CONFIGENTRY']._serialized_end=1669 - _globals['_LOGGINGCONFIG_COUCHBASECOLUMNARDESTINATION']._serialized_start=1671 - _globals['_LOGGINGCONFIG_COUCHBASECOLUMNARDESTINATION']._serialized_end=1754 - _globals['_FEATURESERVICELIST']._serialized_start=1771 - _globals['_FEATURESERVICELIST']._serialized_end=1844 + _globals['_FEATURESERVICESPEC']._serialized_end=550 + _globals['_FEATURESERVICESPEC_TAGSENTRY']._serialized_start=507 + _globals['_FEATURESERVICESPEC_TAGSENTRY']._serialized_end=550 + _globals['_FEATURESERVICEMETA']._serialized_start=553 + _globals['_FEATURESERVICEMETA']._serialized_end=688 + _globals['_LOGGINGCONFIG']._serialized_start=691 + _globals['_LOGGINGCONFIG']._serialized_end=1796 + _globals['_LOGGINGCONFIG_FILEDESTINATION']._serialized_start=1287 + _globals['_LOGGINGCONFIG_FILEDESTINATION']._serialized_end=1370 + _globals['_LOGGINGCONFIG_BIGQUERYDESTINATION']._serialized_start=1372 + _globals['_LOGGINGCONFIG_BIGQUERYDESTINATION']._serialized_end=1412 + _globals['_LOGGINGCONFIG_REDSHIFTDESTINATION']._serialized_start=1414 + _globals['_LOGGINGCONFIG_REDSHIFTDESTINATION']._serialized_end=1455 + _globals['_LOGGINGCONFIG_ATHENADESTINATION']._serialized_start=1457 + _globals['_LOGGINGCONFIG_ATHENADESTINATION']._serialized_end=1496 + _globals['_LOGGINGCONFIG_SNOWFLAKEDESTINATION']._serialized_start=1498 + _globals['_LOGGINGCONFIG_SNOWFLAKEDESTINATION']._serialized_end=1540 + _globals['_LOGGINGCONFIG_CUSTOMDESTINATION']._serialized_start=1543 + _globals['_LOGGINGCONFIG_CUSTOMDESTINATION']._serialized_end=1696 + _globals['_LOGGINGCONFIG_CUSTOMDESTINATION_CONFIGENTRY']._serialized_start=1651 + _globals['_LOGGINGCONFIG_CUSTOMDESTINATION_CONFIGENTRY']._serialized_end=1696 + _globals['_LOGGINGCONFIG_COUCHBASECOLUMNARDESTINATION']._serialized_start=1698 + _globals['_LOGGINGCONFIG_COUCHBASECOLUMNARDESTINATION']._serialized_end=1781 + _globals['_FEATURESERVICELIST']._serialized_start=1798 + _globals['_FEATURESERVICELIST']._serialized_end=1871 # @@protoc_insertion_point(module_scope) diff --git a/sdk/python/feast/protos/feast/core/FeatureService_pb2.pyi b/sdk/python/feast/protos/feast/core/FeatureService_pb2.pyi index 6d5879e52cb..a3e6158ad91 100644 --- a/sdk/python/feast/protos/feast/core/FeatureService_pb2.pyi +++ b/sdk/python/feast/protos/feast/core/FeatureService_pb2.pyi @@ -65,6 +65,7 @@ class FeatureServiceSpec(google.protobuf.message.Message): DESCRIPTION_FIELD_NUMBER: builtins.int OWNER_FIELD_NUMBER: builtins.int LOGGING_CONFIG_FIELD_NUMBER: builtins.int + PRECOMPUTE_ONLINE_FIELD_NUMBER: builtins.int name: builtins.str """Name of the Feature Service. Must be unique. Not updated.""" project: builtins.str @@ -84,6 +85,10 @@ class FeatureServiceSpec(google.protobuf.message.Message): @property def logging_config(self) -> global___LoggingConfig: """(optional) if provided logging will be enabled for this feature service.""" + precompute_online: builtins.bool + """When true, a pre-computed feature vector is maintained per entity for this + service, enabling single-read online retrieval instead of per-feature-view reads. + """ def __init__( self, *, @@ -94,9 +99,10 @@ class FeatureServiceSpec(google.protobuf.message.Message): description: builtins.str = ..., owner: builtins.str = ..., logging_config: global___LoggingConfig | None = ..., + precompute_online: builtins.bool = ..., ) -> None: ... def HasField(self, field_name: typing_extensions.Literal["logging_config", b"logging_config"]) -> builtins.bool: ... - def ClearField(self, field_name: typing_extensions.Literal["description", b"description", "features", b"features", "logging_config", b"logging_config", "name", b"name", "owner", b"owner", "project", b"project", "tags", b"tags"]) -> None: ... + def ClearField(self, field_name: typing_extensions.Literal["description", b"description", "features", b"features", "logging_config", b"logging_config", "name", b"name", "owner", b"owner", "precompute_online", b"precompute_online", "project", b"project", "tags", b"tags"]) -> None: ... global___FeatureServiceSpec = FeatureServiceSpec diff --git a/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.py b/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.py new file mode 100644 index 00000000000..f86f32238cb --- /dev/null +++ b/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.py @@ -0,0 +1,31 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: feast/core/PrecomputedFeatureVector.proto +# Protobuf Python Version: 4.25.1 +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2 +from feast.protos.feast.types import Value_pb2 as feast_dot_types_dot_Value__pb2 + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n)feast/core/PrecomputedFeatureVector.proto\x12\nfeast.core\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x17\x66\x65\x61st/types/Value.proto\"\xc2\x01\n\x18PrecomputedFeatureVector\x12\x15\n\rfeature_names\x18\x01 \x03(\t\x12\"\n\x06values\x18\x02 \x03(\x0b\x32\x12.feast.types.Value\x12\x37\n\rfv_timestamps\x18\x03 \x03(\x0b\x32 .feast.core.FeatureViewTimestamp\x12\x32\n\x0eprecomputed_at\x18\x04 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\"f\n\x14\x46\x65\x61tureViewTimestamp\x12\x19\n\x11\x66\x65\x61ture_view_name\x18\x01 \x01(\t\x12\x33\n\x0f\x65vent_timestamp\x18\x02 \x01(\x0b\x32\x1a.google.protobuf.TimestampBb\n\x10\x66\x65\x61st.proto.coreB\x1dPrecomputedFeatureVectorProtoZ/github.com/feast-dev/feast/go/protos/feast/coreb\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'feast.core.PrecomputedFeatureVector_pb2', _globals) +if _descriptor._USE_C_DESCRIPTORS == False: + _globals['DESCRIPTOR']._options = None + _globals['DESCRIPTOR']._serialized_options = b'\n\020feast.proto.coreB\035PrecomputedFeatureVectorProtoZ/github.com/feast-dev/feast/go/protos/feast/core' + _globals['_PRECOMPUTEDFEATUREVECTOR']._serialized_start=116 + _globals['_PRECOMPUTEDFEATUREVECTOR']._serialized_end=310 + _globals['_FEATUREVIEWTIMESTAMP']._serialized_start=312 + _globals['_FEATUREVIEWTIMESTAMP']._serialized_end=414 +# @@protoc_insertion_point(module_scope) diff --git a/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.pyi b/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.pyi new file mode 100644 index 00000000000..919f1b59973 --- /dev/null +++ b/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.pyi @@ -0,0 +1,76 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" +import builtins +import collections.abc +import feast.types.Value_pb2 +import google.protobuf.descriptor +import google.protobuf.internal.containers +import google.protobuf.message +import google.protobuf.timestamp_pb2 +import sys + +if sys.version_info >= (3, 8): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class PrecomputedFeatureVector(google.protobuf.message.Message): + """A pre-computed feature vector stores all features for a FeatureService + as a single serialized blob per entity, enabling O(1) online retrieval. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FEATURE_NAMES_FIELD_NUMBER: builtins.int + VALUES_FIELD_NUMBER: builtins.int + FV_TIMESTAMPS_FIELD_NUMBER: builtins.int + PRECOMPUTED_AT_FIELD_NUMBER: builtins.int + @property + def feature_names(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: + """Fully-qualified feature names in deterministic order, e.g. "fv1__feat1".""" + @property + def values(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[feast.types.Value_pb2.Value]: + """Feature values in the same order as feature_names.""" + @property + def fv_timestamps(self) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[global___FeatureViewTimestamp]: + """Per-feature-view event timestamps for TTL enforcement at read time.""" + @property + def precomputed_at(self) -> google.protobuf.timestamp_pb2.Timestamp: + """Wall-clock time when this vector was assembled.""" + def __init__( + self, + *, + feature_names: collections.abc.Iterable[builtins.str] | None = ..., + values: collections.abc.Iterable[feast.types.Value_pb2.Value] | None = ..., + fv_timestamps: collections.abc.Iterable[global___FeatureViewTimestamp] | None = ..., + precomputed_at: google.protobuf.timestamp_pb2.Timestamp | None = ..., + ) -> None: ... + def HasField(self, field_name: typing_extensions.Literal["precomputed_at", b"precomputed_at"]) -> builtins.bool: ... + def ClearField(self, field_name: typing_extensions.Literal["feature_names", b"feature_names", "fv_timestamps", b"fv_timestamps", "precomputed_at", b"precomputed_at", "values", b"values"]) -> None: ... + +global___PrecomputedFeatureVector = PrecomputedFeatureVector + +class FeatureViewTimestamp(google.protobuf.message.Message): + """Event timestamp associated with a specific feature view within the vector.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FEATURE_VIEW_NAME_FIELD_NUMBER: builtins.int + EVENT_TIMESTAMP_FIELD_NUMBER: builtins.int + feature_view_name: builtins.str + @property + def event_timestamp(self) -> google.protobuf.timestamp_pb2.Timestamp: ... + def __init__( + self, + *, + feature_view_name: builtins.str = ..., + event_timestamp: google.protobuf.timestamp_pb2.Timestamp | None = ..., + ) -> None: ... + def HasField(self, field_name: typing_extensions.Literal["event_timestamp", b"event_timestamp"]) -> builtins.bool: ... + def ClearField(self, field_name: typing_extensions.Literal["event_timestamp", b"event_timestamp", "feature_view_name", b"feature_view_name"]) -> None: ... + +global___FeatureViewTimestamp = FeatureViewTimestamp diff --git a/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2_grpc.py b/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2_grpc.py new file mode 100644 index 00000000000..2daafffebfc --- /dev/null +++ b/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2_grpc.py @@ -0,0 +1,4 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" +import grpc + diff --git a/sdk/python/tests/unit/test_precomputed_feature_vectors.py b/sdk/python/tests/unit/test_precomputed_feature_vectors.py new file mode 100644 index 00000000000..80b0b83f53f --- /dev/null +++ b/sdk/python/tests/unit/test_precomputed_feature_vectors.py @@ -0,0 +1,933 @@ +"""Tests for pre-computed feature vectors (issue #6185).""" + +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock + +import pytest +from google.protobuf.timestamp_pb2 import Timestamp + +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.on_demand_feature_view import OnDemandFeatureView +from feast.protos.feast.core.PrecomputedFeatureVector_pb2 import ( + FeatureViewTimestamp, + PrecomputedFeatureVector, +) +from feast.protos.feast.serving.ServingService_pb2 import ( + FieldStatus, + GetOnlineFeaturesResponse, +) +from feast.protos.feast.types.Value_pb2 import Value as ValueProto +from feast.types import Float32, Int64 + + +def _make_file_source(name="src"): + return FileSource(name=name, path="test.parquet") + + +def _make_fv(name, features, source=None, ttl=None): + return FeatureView( + name=name, + entities=[], + schema=features, + source=source or _make_file_source(f"{name}_src"), + ttl=ttl or timedelta(0), + ) + + +def _make_timestamp(dt=None): + ts = Timestamp() + ts.FromDatetime(dt or datetime.now(tz=timezone.utc)) + return ts + + +def _make_vector(feature_names, values, fv_timestamps=None, precomputed_at=None): + return PrecomputedFeatureVector( + feature_names=feature_names, + values=values, + fv_timestamps=fv_timestamps or [], + precomputed_at=precomputed_at or _make_timestamp(), + ) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Proto round-trip +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestPrecomputedFeatureVectorProto: + def test_serialize_deserialize(self): + now = _make_timestamp() + fv_ts = FeatureViewTimestamp(feature_view_name="driver_fv", event_timestamp=now) + vector = _make_vector( + feature_names=["driver_fv__trips", "driver_fv__rating"], + values=[ValueProto(int64_val=100), ValueProto(float_val=4.5)], + fv_timestamps=[fv_ts], + precomputed_at=now, + ) + + blob = vector.SerializeToString() + parsed = PrecomputedFeatureVector() + parsed.ParseFromString(blob) + + assert list(parsed.feature_names) == ["driver_fv__trips", "driver_fv__rating"] + assert parsed.values[0].int64_val == 100 + assert parsed.values[1].float_val == pytest.approx(4.5) + assert parsed.fv_timestamps[0].feature_view_name == "driver_fv" + + def test_empty_vector(self): + vector = PrecomputedFeatureVector() + blob = vector.SerializeToString() + parsed = PrecomputedFeatureVector() + parsed.ParseFromString(blob) + assert list(parsed.feature_names) == [] + assert list(parsed.values) == [] + + def test_multiple_fv_timestamps(self): + ts1 = _make_timestamp(datetime(2025, 1, 1, tzinfo=timezone.utc)) + ts2 = _make_timestamp(datetime(2025, 6, 1, tzinfo=timezone.utc)) + + vector = _make_vector( + feature_names=["fv1__a", "fv2__b"], + values=[ValueProto(float_val=1.0), ValueProto(float_val=2.0)], + fv_timestamps=[ + FeatureViewTimestamp(feature_view_name="fv1", event_timestamp=ts1), + FeatureViewTimestamp(feature_view_name="fv2", event_timestamp=ts2), + ], + ) + blob = vector.SerializeToString() + parsed = PrecomputedFeatureVector() + parsed.ParseFromString(blob) + + assert len(parsed.fv_timestamps) == 2 + assert parsed.fv_timestamps[0].feature_view_name == "fv1" + assert parsed.fv_timestamps[1].feature_view_name == "fv2" + assert parsed.fv_timestamps[0].event_timestamp.seconds == ts1.seconds + assert parsed.fv_timestamps[1].event_timestamp.seconds == ts2.seconds + + def test_null_values_preserved(self): + vector = _make_vector( + feature_names=["fv1__a", "fv1__b"], + values=[ValueProto(float_val=1.0), ValueProto()], + ) + blob = vector.SerializeToString() + parsed = PrecomputedFeatureVector() + parsed.ParseFromString(blob) + + assert parsed.values[0].float_val == pytest.approx(1.0) + assert not parsed.values[1].HasField("val") + + def test_large_vector(self): + n = 100 + names = [f"fv__feat_{i}" for i in range(n)] + vals = [ValueProto(float_val=float(i)) for i in range(n)] + vector = _make_vector(feature_names=names, values=vals) + blob = vector.SerializeToString() + parsed = PrecomputedFeatureVector() + parsed.ParseFromString(blob) + + assert len(parsed.feature_names) == n + assert len(parsed.values) == n + assert parsed.values[99].float_val == pytest.approx(99.0) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# FeatureService precompute_online flag +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestFeatureServicePrecomputeOnline: + def test_default_is_false(self): + fs = FeatureService(name="svc", features=[]) + assert fs.precompute_online is False + + def test_set_true(self): + fs = FeatureService(name="svc", features=[], precompute_online=True) + assert fs.precompute_online is True + + def test_to_proto_roundtrip_true(self): + fv = _make_fv("fv1", [Field(name="f1", dtype=Float32)]) + fs = FeatureService(name="svc", features=[fv], precompute_online=True) + proto = fs.to_proto() + assert proto.spec.precompute_online is True + restored = FeatureService.from_proto(proto) + assert restored.precompute_online is True + + def test_to_proto_roundtrip_false(self): + fs = FeatureService(name="svc", features=[], precompute_online=False) + proto = fs.to_proto() + assert proto.spec.precompute_online is False + restored = FeatureService.from_proto(proto) + assert restored.precompute_online is False + + def test_equality_includes_precompute(self): + fv = _make_fv("fv1", [Field(name="f1", dtype=Float32)]) + fs1 = FeatureService(name="svc", features=[fv], precompute_online=True) + fs2 = FeatureService(name="svc", features=[fv], precompute_online=False) + assert fs1 != fs2 + + def test_equality_same_precompute(self): + fv = _make_fv("fv1", [Field(name="f1", dtype=Float32)]) + fs1 = FeatureService(name="svc", features=[fv], precompute_online=True) + fs2 = FeatureService(name="svc", features=[fv], precompute_online=True) + assert fs1 == fs2 + + def test_proto_preserves_other_fields(self): + fv = _make_fv("fv1", [Field(name="f1", dtype=Float32)]) + fs = FeatureService( + name="svc", + features=[fv], + precompute_online=True, + description="test desc", + owner="owner@test.com", + tags={"env": "prod"}, + ) + proto = fs.to_proto() + restored = FeatureService.from_proto(proto) + + assert restored.precompute_online is True + assert restored.description == "test desc" + assert restored.owner == "owner@test.com" + assert restored.tags == {"env": "prod"} + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Validation +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestFeatureServiceValidation: + def test_validate_passes_without_precompute(self): + fs = FeatureService(name="svc", features=[], precompute_online=False) + fs.validate() + + def test_validate_passes_precompute_regular_fv(self): + fv = _make_fv("fv1", [Field(name="f1", dtype=Float32)]) + fs = FeatureService(name="svc", features=[fv], precompute_online=True) + fs.validate() + + def test_validate_rejects_odfv_without_write_to_online(self): + fv = _make_fv("fv1", [Field(name="f1", dtype=Float32)]) + + odfv = MagicMock(spec=OnDemandFeatureView) + odfv.name = "odfv1" + odfv.write_to_online_store = False + + fs = FeatureService(name="svc", features=[fv], precompute_online=True) + fs._features = [fv, odfv] + + with pytest.raises(ValueError, match="write_to_online_store=False"): + fs.validate() + + def test_validate_passes_odfv_with_write_to_online(self): + fv = _make_fv("fv1", [Field(name="f1", dtype=Float32)]) + + odfv = MagicMock(spec=OnDemandFeatureView) + odfv.name = "odfv1" + odfv.write_to_online_store = True + + fs = FeatureService(name="svc", features=[fv], precompute_online=True) + fs._features = [fv, odfv] + + fs.validate() + + def test_validate_rejects_multiple_odfvs_without_write(self): + odfv1 = MagicMock(spec=OnDemandFeatureView) + odfv1.name = "odfv1" + odfv1.write_to_online_store = False + + odfv2 = MagicMock(spec=OnDemandFeatureView) + odfv2.name = "odfv2" + odfv2.write_to_online_store = False + + fs = FeatureService(name="svc", features=[], precompute_online=True) + fs._features = [odfv1, odfv2] + + with pytest.raises(ValueError, match="odfv1"): + fs.validate() + + def test_validate_error_message_includes_service_and_odfv_name(self): + odfv = MagicMock(spec=OnDemandFeatureView) + odfv.name = "my_transform" + odfv.write_to_online_store = False + + fs = FeatureService(name="my_service", features=[], precompute_online=True) + fs._features = [odfv] + + with pytest.raises(ValueError) as exc_info: + fs.validate() + + assert "my_service" in str(exc_info.value) + assert "my_transform" in str(exc_info.value) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Schema mismatch detection (fast path fallback) +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestSchemaMismatchFallback: + def _fast_path(self, blobs, expected_names, grouped_refs=None, num_rows=None): + from feast.infra.online_stores.online_store import OnlineStore + + response = GetOnlineFeaturesResponse(results=[]) + result = OnlineStore._try_precomputed_fast_path( + blobs=blobs, + expected_feature_names=expected_names, + online_features_response=response, + full_feature_names=True, + num_rows=num_rows or len(blobs), + grouped_refs=grouped_refs or [], + registry=MagicMock(), + project="test", + ) + return result, response + + def test_mismatch_returns_false(self): + vector = _make_vector( + feature_names=["fv1__f1", "fv1__f2"], + values=[ValueProto(float_val=1.0), ValueProto(float_val=2.0)], + ) + ok, _ = self._fast_path([vector.SerializeToString()], ["fv1__f1", "fv1__f3"]) + assert ok is False + + def test_none_blob_returns_false(self): + ok, _ = self._fast_path([None], ["fv1__f1"]) + assert ok is False + + def test_matching_schema_returns_true(self): + vector = _make_vector( + feature_names=["fv1__f1"], + values=[ValueProto(float_val=42.0)], + ) + ok, response = self._fast_path([vector.SerializeToString()], ["fv1__f1"]) + assert ok is True + assert len(response.results) == 1 + assert response.results[0].values[0].float_val == pytest.approx(42.0) + + def test_extra_feature_in_expected_returns_false(self): + vector = _make_vector( + feature_names=["fv1__f1"], + values=[ValueProto(float_val=1.0)], + ) + ok, _ = self._fast_path([vector.SerializeToString()], ["fv1__f1", "fv1__f2"]) + assert ok is False + + def test_fewer_features_in_expected_returns_false(self): + vector = _make_vector( + feature_names=["fv1__f1", "fv1__f2"], + values=[ValueProto(float_val=1.0), ValueProto(float_val=2.0)], + ) + ok, _ = self._fast_path([vector.SerializeToString()], ["fv1__f1"]) + assert ok is False + + def test_reordered_features_succeeds_with_correct_values(self): + vector = _make_vector( + feature_names=["fv1__f1", "fv1__f2"], + values=[ValueProto(float_val=1.0), ValueProto(float_val=2.0)], + ) + ok, response = self._fast_path( + [vector.SerializeToString()], ["fv1__f2", "fv1__f1"] + ) + assert ok is True + assert response.results[0].values[0].float_val == pytest.approx(2.0) + assert response.results[1].values[0].float_val == pytest.approx(1.0) + + def test_all_none_blobs_returns_false(self): + ok, _ = self._fast_path([None, None, None], ["fv1__f1"], num_rows=3) + assert ok is False + + def test_mixed_none_and_valid_returns_false(self): + vector = _make_vector( + feature_names=["fv1__f1"], + values=[ValueProto(float_val=1.0)], + ) + ok, _ = self._fast_path( + [vector.SerializeToString(), None], ["fv1__f1"], num_rows=2 + ) + assert ok is False + + def test_schema_mismatch_in_second_entity_returns_false(self): + vec1 = _make_vector( + feature_names=["fv1__f1"], values=[ValueProto(float_val=1.0)] + ) + vec2 = _make_vector( + feature_names=["fv1__f_WRONG"], values=[ValueProto(float_val=2.0)] + ) + ok, _ = self._fast_path( + [vec1.SerializeToString(), vec2.SerializeToString()], + ["fv1__f1"], + num_rows=2, + ) + assert ok is False + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Fast path: multi-entity response building +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestFastPathMultiEntity: + def _fast_path(self, blobs, expected_names, grouped_refs=None, num_rows=None): + from feast.infra.online_stores.online_store import OnlineStore + + response = GetOnlineFeaturesResponse(results=[]) + result = OnlineStore._try_precomputed_fast_path( + blobs=blobs, + expected_feature_names=expected_names, + online_features_response=response, + full_feature_names=True, + num_rows=num_rows or len(blobs), + grouped_refs=grouped_refs or [], + registry=MagicMock(), + project="test", + ) + return result, response + + def test_multiple_entities_all_present(self): + vec1 = _make_vector( + feature_names=["fv1__a", "fv1__b"], + values=[ValueProto(float_val=1.0), ValueProto(float_val=10.0)], + ) + vec2 = _make_vector( + feature_names=["fv1__a", "fv1__b"], + values=[ValueProto(float_val=2.0), ValueProto(float_val=20.0)], + ) + vec3 = _make_vector( + feature_names=["fv1__a", "fv1__b"], + values=[ValueProto(float_val=3.0), ValueProto(float_val=30.0)], + ) + + ok, response = self._fast_path( + [ + vec1.SerializeToString(), + vec2.SerializeToString(), + vec3.SerializeToString(), + ], + ["fv1__a", "fv1__b"], + num_rows=3, + ) + assert ok is True + assert len(response.results) == 2 + + a_values = [v.float_val for v in response.results[0].values] + b_values = [v.float_val for v in response.results[1].values] + + assert a_values == pytest.approx([1.0, 2.0, 3.0]) + assert b_values == pytest.approx([10.0, 20.0, 30.0]) + + def test_statuses_are_present(self): + vec = _make_vector( + feature_names=["fv1__a"], + values=[ValueProto(float_val=1.0)], + ) + ok, response = self._fast_path( + [vec.SerializeToString()], ["fv1__a"], num_rows=1 + ) + assert ok is True + assert response.results[0].statuses[0] == FieldStatus.PRESENT + + def test_feature_names_in_metadata(self): + vec = _make_vector( + feature_names=["fv1__x", "fv2__y"], + values=[ValueProto(float_val=1.0), ValueProto(float_val=2.0)], + ) + ok, response = self._fast_path([vec.SerializeToString()], ["fv1__x", "fv2__y"]) + assert ok is True + assert list(response.metadata.feature_names.val) == ["fv1__x", "fv2__y"] + + +# ═══════════════════════════════════════════════════════════════════════════════ +# TTL enforcement in fast path +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestFastPathTTLEnforcement: + def _fast_path_with_ttl(self, fv_event_dt, fv_ttl_seconds, fv_name="fv1"): + from feast.infra.online_stores.online_store import OnlineStore + + event_ts = _make_timestamp(fv_event_dt) + fv_ts = FeatureViewTimestamp( + feature_view_name=fv_name, event_timestamp=event_ts + ) + vector = _make_vector( + feature_names=[f"{fv_name}__f1"], + values=[ValueProto(float_val=1.0)], + fv_timestamps=[fv_ts], + ) + + fv = _make_fv( + fv_name, + [Field(name="f1", dtype=Float32)], + ttl=timedelta(seconds=fv_ttl_seconds), + ) + grouped_refs = [(fv, ["f1"])] + + response = GetOnlineFeaturesResponse(results=[]) + ok = OnlineStore._try_precomputed_fast_path( + blobs=[vector.SerializeToString()], + expected_feature_names=[f"{fv_name}__f1"], + online_features_response=response, + full_feature_names=True, + num_rows=1, + grouped_refs=grouped_refs, + registry=MagicMock(), + project="test", + ) + return ok, response + + def test_fresh_data_is_present(self): + now = datetime.now(tz=timezone.utc) + ok, response = self._fast_path_with_ttl(now, fv_ttl_seconds=3600) + assert ok is True + assert response.results[0].statuses[0] == FieldStatus.PRESENT + + def test_expired_data_is_outside_max_age(self): + old = datetime.now(tz=timezone.utc) - timedelta(hours=2) + ok, response = self._fast_path_with_ttl(old, fv_ttl_seconds=3600) + assert ok is True + assert response.results[0].statuses[0] == FieldStatus.OUTSIDE_MAX_AGE + + def test_zero_ttl_means_no_expiry(self): + old = datetime.now(tz=timezone.utc) - timedelta(days=365) + fv = _make_fv( + "fv1", + [Field(name="f1", dtype=Float32)], + ttl=timedelta(0), + ) + + event_ts = _make_timestamp(old) + fv_ts = FeatureViewTimestamp(feature_view_name="fv1", event_timestamp=event_ts) + vector = _make_vector( + feature_names=["fv1__f1"], + values=[ValueProto(float_val=1.0)], + fv_timestamps=[fv_ts], + ) + + from feast.infra.online_stores.online_store import OnlineStore + + response = GetOnlineFeaturesResponse(results=[]) + ok = OnlineStore._try_precomputed_fast_path( + blobs=[vector.SerializeToString()], + expected_feature_names=["fv1__f1"], + online_features_response=response, + full_feature_names=True, + num_rows=1, + grouped_refs=[(fv, ["f1"])], + registry=MagicMock(), + project="test", + ) + assert ok is True + assert response.results[0].statuses[0] == FieldStatus.PRESENT + + def test_mixed_ttl_some_expired_some_fresh(self): + from feast.infra.online_stores.online_store import OnlineStore + + now = datetime.now(tz=timezone.utc) + fresh_ts = _make_timestamp(now) + stale_ts = _make_timestamp(now - timedelta(hours=5)) + + vector = _make_vector( + feature_names=["fv1__f1", "fv2__f2"], + values=[ValueProto(float_val=1.0), ValueProto(float_val=2.0)], + fv_timestamps=[ + FeatureViewTimestamp(feature_view_name="fv1", event_timestamp=fresh_ts), + FeatureViewTimestamp(feature_view_name="fv2", event_timestamp=stale_ts), + ], + ) + + fv1 = _make_fv("fv1", [Field(name="f1", dtype=Float32)], ttl=timedelta(hours=1)) + fv2 = _make_fv("fv2", [Field(name="f2", dtype=Float32)], ttl=timedelta(hours=1)) + + response = GetOnlineFeaturesResponse(results=[]) + ok = OnlineStore._try_precomputed_fast_path( + blobs=[vector.SerializeToString()], + expected_feature_names=["fv1__f1", "fv2__f2"], + online_features_response=response, + full_feature_names=True, + num_rows=1, + grouped_refs=[(fv1, ["f1"]), (fv2, ["f2"])], + registry=MagicMock(), + project="test", + ) + assert ok is True + assert response.results[0].statuses[0] == FieldStatus.PRESENT + assert response.results[1].statuses[0] == FieldStatus.OUTSIDE_MAX_AGE + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Expected feature names computation +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestComputeExpectedFeatureNames: + def _compute(self, grouped_refs, full_feature_names): + from feast.infra.online_stores.online_store import OnlineStore + + return OnlineStore._compute_expected_feature_names( + grouped_refs, full_feature_names + ) + + def test_full_feature_names(self): + fv = _make_fv("driver_fv", [Field(name="trips", dtype=Int64)]) + assert self._compute([(fv, ["trips"])], True) == ["driver_fv__trips"] + + def test_short_feature_names(self): + fv = _make_fv("driver_fv", [Field(name="trips", dtype=Int64)]) + assert self._compute([(fv, ["trips"])], False) == ["trips"] + + def test_skips_ts_keys(self): + fv = _make_fv("driver_fv", [Field(name="trips", dtype=Int64)]) + result = self._compute([(fv, ["trips", "_ts:driver_fv"])], True) + assert result == ["driver_fv__trips"] + + def test_multiple_fvs(self): + fv1 = _make_fv("fv1", [Field(name="a", dtype=Float32)]) + fv2 = _make_fv("fv2", [Field(name="b", dtype=Float32)]) + result = self._compute([(fv1, ["a"]), (fv2, ["b"])], True) + assert result == ["fv1__a", "fv2__b"] + + def test_multiple_features_per_fv(self): + fv = _make_fv( + "fv1", + [ + Field(name="a", dtype=Float32), + Field(name="b", dtype=Float32), + Field(name="c", dtype=Float32), + ], + ) + result = self._compute([(fv, ["a", "b", "c"])], True) + assert result == ["fv1__a", "fv1__b", "fv1__c"] + + def test_empty_grouped_refs(self): + assert self._compute([], True) == [] + + def test_only_ts_keys(self): + fv = _make_fv("fv1", [Field(name="a", dtype=Float32)]) + result = self._compute([(fv, ["_ts:fv1"])], True) + assert result == [] + + +# ═══════════════════════════════════════════════════════════════════════════════ +# OnlineStore base class default implementations +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestOnlineStoreBaseDefaults: + def _make_dummy_store(self, online_read_fn=None, online_write_batch_fn=None): + from feast.infra.online_stores.online_store import OnlineStore + + class DummyStore(OnlineStore): + def __init__(self): + self.written = [] + + def online_write_batch(self, config, table, data, progress): + self.written.append((table.name, data)) + if online_write_batch_fn: + online_write_batch_fn(config, table, data, progress) + + def online_read(self, config, table, entity_keys, requested_features=None): + if online_read_fn: + return online_read_fn( + config, table, entity_keys, requested_features + ) + raise RuntimeError("not implemented") + + def update(self, *args, **kwargs): + pass + + def teardown(self, *args, **kwargs): + pass + + return DummyStore() + + def test_read_precomputed_returns_none_on_exception(self): + store = self._make_dummy_store() + config = MagicMock() + entity_keys = [MagicMock(), MagicMock()] + + result = store.read_precomputed_vectors( + config, "my_service", "project", entity_keys + ) + assert result == [None, None] + + def test_read_precomputed_returns_none_for_missing_entities(self): + def online_read_fn(config, table, entity_keys, requested_features): + return [(None, None) for _ in entity_keys] + + store = self._make_dummy_store(online_read_fn=online_read_fn) + result = store.read_precomputed_vectors( + MagicMock(), "svc", "proj", [MagicMock()] + ) + assert result == [None] + + def test_read_precomputed_returns_bytes_for_present_entities(self): + test_bytes = b"test_vector_data" + + def online_read_fn(config, table, entity_keys, requested_features): + return [ + ( + datetime.now(tz=timezone.utc), + {"vector": ValueProto(bytes_val=test_bytes)}, + ) + for _ in entity_keys + ] + + store = self._make_dummy_store(online_read_fn=online_read_fn) + result = store.read_precomputed_vectors( + MagicMock(), "svc", "proj", [MagicMock()] + ) + assert result == [test_bytes] + + def test_write_precomputed_delegates_to_online_write_batch(self): + store = self._make_dummy_store() + entity_key = MagicMock() + vector_bytes = b"blob_data" + + store.write_precomputed_vector( + MagicMock(), "my_svc", "proj", entity_key, vector_bytes + ) + assert len(store.written) == 1 + table_name, data = store.written[0] + assert table_name == "__precomputed__my_svc" + assert len(data) == 1 + assert data[0][1]["vector"].bytes_val == vector_bytes + + def test_write_then_read_roundtrip(self): + stored = {} + + def online_write_fn(config, table, data, progress): + for ek, vals, ts, _ in data: + stored[id(ek)] = (ts, vals) + + def online_read_fn(config, table, entity_keys, requested_features): + results = [] + for ek in entity_keys: + if id(ek) in stored: + ts, vals = stored[id(ek)] + results.append((ts, vals)) + else: + results.append((None, None)) + return results + + store = self._make_dummy_store( + online_read_fn=online_read_fn, + online_write_batch_fn=online_write_fn, + ) + + entity_key = MagicMock() + vector_bytes = b"test_roundtrip" + + store.write_precomputed_vector( + MagicMock(), "svc", "proj", entity_key, vector_bytes + ) + result = store.read_precomputed_vectors( + MagicMock(), "svc", "proj", [entity_key] + ) + assert result == [vector_bytes] + + def test_read_returns_none_for_non_bytes_val(self): + def online_read_fn(config, table, entity_keys, requested_features): + return [ + ( + datetime.now(tz=timezone.utc), + {"vector": ValueProto(string_val="not bytes")}, + ) + for _ in entity_keys + ] + + store = self._make_dummy_store(online_read_fn=online_read_fn) + result = store.read_precomputed_vectors( + MagicMock(), "svc", "proj", [MagicMock()] + ) + assert result == [None] + + def test_read_returns_none_when_no_vector_key(self): + def online_read_fn(config, table, entity_keys, requested_features): + return [ + ( + datetime.now(tz=timezone.utc), + {"other_field": ValueProto(float_val=1.0)}, + ) + for _ in entity_keys + ] + + store = self._make_dummy_store(online_read_fn=online_read_fn) + result = store.read_precomputed_vectors( + MagicMock(), "svc", "proj", [MagicMock()] + ) + assert result == [None] + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Materialization hook: _precompute_affected_services +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestPrecomputeAffectedServices: + def _make_mock_store(self, services): + store = MagicMock() + store.registry.list_feature_services.return_value = services + store.project = "test_project" + store._precompute_affected_services = lambda fv_names: type( + store + )._precompute_affected_services(store, fv_names) + + from feast.feature_store import FeatureStore + + store._precompute_affected_services = ( + FeatureStore._precompute_affected_services.__get__(store) + ) + return store + + def test_triggers_precompute_for_matching_service(self): + proj = MagicMock() + proj.name = "fv1" + svc = MagicMock() + svc.precompute_online = True + svc.feature_view_projections = [proj] + svc.name = "my_service" + + store = self._make_mock_store([svc]) + store._precompute_affected_services(["fv1"]) + store.precompute_feature_service.assert_called_once_with("my_service") + + def test_skips_non_precompute_services(self): + svc = MagicMock() + svc.precompute_online = False + svc.name = "svc" + + store = self._make_mock_store([svc]) + store._precompute_affected_services(["fv1"]) + store.precompute_feature_service.assert_not_called() + + def test_skips_when_no_fvs_match(self): + proj = MagicMock() + proj.name = "fv_other" + svc = MagicMock() + svc.precompute_online = True + svc.feature_view_projections = [proj] + svc.name = "svc" + + store = self._make_mock_store([svc]) + store._precompute_affected_services(["fv1"]) + store.precompute_feature_service.assert_not_called() + + def test_handles_registry_exception_gracefully(self): + store = MagicMock() + store.registry.list_feature_services.side_effect = RuntimeError("oops") + store.project = "test" + + from feast.feature_store import FeatureStore + + FeatureStore._precompute_affected_services(store, ["fv1"]) + + def test_handles_precompute_exception_gracefully(self): + proj = MagicMock() + proj.name = "fv1" + svc = MagicMock() + svc.precompute_online = True + svc.feature_view_projections = [proj] + svc.name = "svc" + + store = self._make_mock_store([svc]) + store.precompute_feature_service.side_effect = RuntimeError("fail") + + store._precompute_affected_services(["fv1"]) + + def test_multiple_services_all_triggered(self): + proj1 = MagicMock() + proj1.name = "fv1" + svc1 = MagicMock() + svc1.precompute_online = True + svc1.feature_view_projections = [proj1] + svc1.name = "svc1" + + proj2 = MagicMock() + proj2.name = "fv1" + svc2 = MagicMock() + svc2.precompute_online = True + svc2.feature_view_projections = [proj2] + svc2.name = "svc2" + + store = self._make_mock_store([svc1, svc2]) + store._precompute_affected_services(["fv1"]) + + assert store.precompute_feature_service.call_count == 2 + + +# ═══════════════════════════════════════════════════════════════════════════════ +# Push hook: _precompute_for_push +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestPrecomputeForPush: + def _make_mock_store(self, services): + store = MagicMock() + store.registry.list_feature_services.return_value = services + store.project = "test_project" + + from feast.feature_store import FeatureStore + + store._precompute_for_push = FeatureStore._precompute_for_push.__get__(store) + return store + + def test_triggers_for_matching_fv(self): + proj = MagicMock() + proj.name = "pushed_fv" + svc = MagicMock() + svc.precompute_online = True + svc.feature_view_projections = [proj] + svc.name = "svc" + + store = self._make_mock_store([svc]) + store._precompute_for_push("pushed_fv", MagicMock()) + store.precompute_feature_service.assert_called_once_with("svc") + + def test_skips_non_matching_fv(self): + proj = MagicMock() + proj.name = "other_fv" + svc = MagicMock() + svc.precompute_online = True + svc.feature_view_projections = [proj] + svc.name = "svc" + + store = self._make_mock_store([svc]) + store._precompute_for_push("pushed_fv", MagicMock()) + store.precompute_feature_service.assert_not_called() + + def test_skips_non_precompute_service(self): + proj = MagicMock() + proj.name = "pushed_fv" + svc = MagicMock() + svc.precompute_online = False + svc.feature_view_projections = [proj] + svc.name = "svc" + + store = self._make_mock_store([svc]) + store._precompute_for_push("pushed_fv", MagicMock()) + store.precompute_feature_service.assert_not_called() + + def test_handles_exception_gracefully(self): + proj = MagicMock() + proj.name = "fv" + svc = MagicMock() + svc.precompute_online = True + svc.feature_view_projections = [proj] + svc.name = "svc" + + store = self._make_mock_store([svc]) + store.precompute_feature_service.side_effect = RuntimeError("fail") + + store._precompute_for_push("fv", MagicMock()) + + def test_handles_registry_exception_gracefully(self): + store = MagicMock() + store.registry.list_feature_services.side_effect = RuntimeError("oops") + store.project = "test" + + from feast.feature_store import FeatureStore + + FeatureStore._precompute_for_push(store, "fv", MagicMock())