diff --git a/protos/feast/core/FeatureService.proto b/protos/feast/core/FeatureService.proto index 557782fe552..ab74e036a6d 100644 --- a/protos/feast/core/FeatureService.proto +++ b/protos/feast/core/FeatureService.proto @@ -42,6 +42,10 @@ message FeatureServiceSpec { // 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; + + // When true, pre-computed vectors store only values and a schema fingerprint, + // not the repeated feature name list. + bool precompute_omit_names = 9; } diff --git a/protos/feast/core/PrecomputedFeatureVector.proto b/protos/feast/core/PrecomputedFeatureVector.proto index 798df0b8d1e..9880dcac963 100644 --- a/protos/feast/core/PrecomputedFeatureVector.proto +++ b/protos/feast/core/PrecomputedFeatureVector.proto @@ -22,6 +22,10 @@ message PrecomputedFeatureVector { // Wall-clock time when this vector was assembled. google.protobuf.Timestamp precomputed_at = 4; + + // Fingerprint of the FeatureService's ordered "__" names. Lets a + // vector omit feature_names and still be validated against the service. + string schema_fingerprint = 5; } // Event timestamp associated with a specific feature view within the vector. diff --git a/sdk/python/feast/feature_service.py b/sdk/python/feast/feature_service.py index 23169ac8f93..b595312dac5 100644 --- a/sdk/python/feast/feature_service.py +++ b/sdk/python/feast/feature_service.py @@ -1,4 +1,5 @@ import copy +import hashlib from datetime import datetime from typing import TYPE_CHECKING, Dict, List, Optional, Union @@ -61,6 +62,7 @@ class FeatureService: logging_config: Optional[LoggingConfig] = None precompute_online: bool = False + precompute_omit_names: bool = False def __init__( self, @@ -72,6 +74,7 @@ def __init__( owner: str = "", logging_config: Optional[LoggingConfig] = None, precompute_online: bool = False, + precompute_omit_names: bool = False, ): """ Creates a FeatureService object. @@ -108,6 +111,7 @@ def __init__( self.last_updated_timestamp = None self.logging_config = logging_config self.precompute_online = precompute_online + self.precompute_omit_names = precompute_omit_names for feature_grouping in self._features: if isinstance(feature_grouping, str): # No registry at construction time; resolved in resolve_pending_refs. @@ -360,6 +364,7 @@ def prepare_for_apply( owner=self.owner, logging_config=self.logging_config, precompute_online=self.precompute_online, + precompute_omit_names=self.precompute_omit_names, ) prepared.created_timestamp = self.created_timestamp prepared.last_updated_timestamp = self.last_updated_timestamp @@ -394,6 +399,20 @@ def __str__(self): def __hash__(self): return hash(self.name) + def precompute_feature_names(self) -> List[str]: + """``__`` names in the order pre-computed vectors store values.""" + return [ + f"{projection.name_to_use()}__{feature.name}" + for projection in self.feature_view_projections + for feature in projection.features + ] + + def precompute_schema_fingerprint(self) -> str: + """Stable digest of :meth:`precompute_feature_names` for nameless vectors.""" + return hashlib.sha256( + "\n".join(self.precompute_feature_names()).encode("utf-8") + ).hexdigest()[:16] + def __eq__(self, other): if not isinstance(other, FeatureService): return False @@ -404,6 +423,7 @@ def __eq__(self, other): or self.tags != other.tags or self.owner != other.owner or self.precompute_online != other.precompute_online + or self.precompute_omit_names != other.precompute_omit_names ): return False @@ -432,6 +452,7 @@ def from_proto(cls, feature_service_proto: FeatureServiceProto): feature_service_proto.spec.logging_config ), precompute_online=feature_service_proto.spec.precompute_online, + precompute_omit_names=feature_service_proto.spec.precompute_omit_names, ) fs.feature_view_projections.extend( [ @@ -484,6 +505,7 @@ def to_proto(self) -> FeatureServiceProto: if self.logging_config else None, precompute_online=self.precompute_online, + precompute_omit_names=self.precompute_omit_names, ) return FeatureServiceProto(spec=spec, meta=meta) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 917454d19ad..a8fa7143883 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -115,6 +115,7 @@ _get_feature_view_vector_field_metadata, _utc_now, ) +from feast.value_type import ValueType from feast.vector_store_utils import feature_view_to_vs_id from feast.version_utils import parse_version @@ -3183,6 +3184,7 @@ def precompute_feature_service( self, feature_service_name: Optional[str] = None, batch_size: int = 1000, + entity_keys: Optional[List[EntityKey]] = None, ) -> int: """Pre-compute feature vectors for one or all FeatureServices. @@ -3193,6 +3195,9 @@ def precompute_feature_service( Works with **all** online store backends (Redis, DynamoDB, PostgreSQL, etc.). + When ``entity_keys`` is given, only those entities are rebuilt and the store + is not scanned, so a push can refresh exactly the rows it wrote. + Returns the total number of entity vectors written. """ from feast.protos.feast.core.PrecomputedFeatureVector_pb2 import ( @@ -3240,53 +3245,56 @@ def precompute_feature_service( # 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 entity_keys is not None: + entity_key_protos = list(entity_keys) + else: + 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 + 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__, ) - 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 + continue if not entity_key_protos: continue @@ -3338,10 +3346,13 @@ def precompute_feature_service( now_ts.FromDatetime(datetime.now(tz=_timezone.utc)) vector = PrecomputedFeatureVector( - feature_names=feature_names, + feature_names=[] + if svc.precompute_omit_names + else feature_names, values=values, fv_timestamps=fv_timestamps, precomputed_at=now_ts, + schema_fingerprint=svc.precompute_schema_fingerprint(), ) online_store.write_precomputed_vector( @@ -3399,9 +3410,16 @@ def _precompute_for_push(self, feature_view_name: str, df: "pd.DataFrame") -> No if not affected: return + fv = self.registry.get_any_feature_view( + feature_view_name, self.project, allow_cache=True + ) + entity_keys = _entity_keys_from_dataframe(fv, df) + if not entity_keys: + return + for svc in affected: try: - self.precompute_feature_service(svc.name) + self.precompute_feature_service(svc.name, entity_keys=entity_keys) except Exception: _logger.warning( "Failed to precompute vectors for service '%s' after push", @@ -5210,3 +5228,27 @@ def _validate_data_sources( UserWarning, stacklevel=2, ) + + +def _entity_keys_from_dataframe(feature_view: Any, df: pd.DataFrame) -> List[EntityKey]: + """Distinct entity keys present in ``df`` for ``feature_view``'s join keys.""" + from feast.type_map import python_values_to_proto_values + + join_keys = list(getattr(feature_view, "join_keys", []) or []) + if not join_keys or any(k not in df.columns for k in join_keys): + return [] + value_types = { + field.name: field.dtype.to_value_type() + for field in getattr(feature_view, "entity_columns", []) + } + unique = df[join_keys].drop_duplicates() + columns = { + k: python_values_to_proto_values( + unique[k].tolist(), value_types.get(k, ValueType.UNKNOWN) + ) + for k in join_keys + } + return [ + EntityKey(join_keys=join_keys, entity_values=[columns[k][i] for k in join_keys]) + for i in range(len(unique)) + ] diff --git a/sdk/python/feast/infra/online_stores/online_store.py b/sdk/python/feast/infra/online_stores/online_store.py index 69d596dc6bf..8527bf94c87 100644 --- a/sdk/python/feast/infra/online_stores/online_store.py +++ b/sdk/python/feast/infra/online_stores/online_store.py @@ -241,6 +241,7 @@ def get_online_features( grouped_refs, registry, project, + feature_service=features, ) if not precomputed_ok: raise RuntimeError( @@ -527,6 +528,7 @@ async def get_online_features_async( grouped_refs, registry, project, + feature_service=features, ) if not precomputed_ok: raise RuntimeError( @@ -612,6 +614,7 @@ def _try_precomputed_fast_path( grouped_refs: List, registry: BaseRegistry, project: str, + feature_service: Optional[Any] = None, ) -> bool: """Build the response from pre-computed vectors. @@ -662,27 +665,65 @@ def _try_precomputed_fast_path( reorder_map: Optional[List[int]] = None vectors: List[PrecomputedFeatureVector] = [] + # A nameless vector is validated by fingerprint and takes its names, in + # stored order, from the FeatureService it was built for. + service_names: Optional[List[str]] = None + service_fingerprint: Optional[str] = None + if feature_service is not None and hasattr( + feature_service, "precompute_schema_fingerprint" + ): + service_names = feature_service.precompute_feature_names() + service_fingerprint = feature_service.precompute_schema_fingerprint() + for blob in blobs: vec = PrecomputedFeatureVector() vec.ParseFromString(blob) # type: ignore[arg-type] vectors.append(vec) + stored_names = list(vec.feature_names) + if not stored_names: + if ( + service_names is None + or vec.schema_fingerprint != service_fingerprint + ): + logger.warning( + "Pre-computed vector fingerprint mismatch at index %d: stored=%s, expected=%s", + len(vectors) - 1, + vec.schema_fingerprint, + service_fingerprint, + ) + return False + stored_names = service_names + if first_stored_names is None: - first_stored_names = list(vec.feature_names) - if set(first_stored_names) != expected_set: + first_stored_names = stored_names + # Vectors always store "__"; the caller's expected + # names are bare unless full_feature_names was requested. + comparable_names = ( + first_stored_names + if full_feature_names + else [n.split("__", 1)[-1] for n in first_stored_names] + ) + if set(comparable_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)} + if comparable_names != expected_feature_names: + name_to_idx = {n: i for i, n in enumerate(comparable_names)} reorder_map = [name_to_idx[n] for n in expected_feature_names] + feat_fv_names = [ + first_stored_names[reorder_map[i] if reorder_map else i].split( + "__", 1 + )[0] + for i in range(n_features) + ] else: if ( - len(vec.feature_names) != n_features - or list(vec.feature_names) != first_stored_names + len(stored_names) != n_features + or stored_names != first_stored_names ): logger.warning( "Pre-computed vector schema varies across entities at index %d", diff --git a/sdk/python/feast/protos/feast/core/FeatureService_pb2.py b/sdk/python/feast/protos/feast/core/FeatureService_pb2.py index 24940b4ca33..37f6739853a 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\"\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') +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\"\xde\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\x12\x1d\n\x15precompute_omit_names\x18\t \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=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 + _globals['_FEATURESERVICESPEC']._serialized_end=581 + _globals['_FEATURESERVICESPEC_TAGSENTRY']._serialized_start=538 + _globals['_FEATURESERVICESPEC_TAGSENTRY']._serialized_end=581 + _globals['_FEATURESERVICEMETA']._serialized_start=584 + _globals['_FEATURESERVICEMETA']._serialized_end=719 + _globals['_LOGGINGCONFIG']._serialized_start=722 + _globals['_LOGGINGCONFIG']._serialized_end=1827 + _globals['_LOGGINGCONFIG_FILEDESTINATION']._serialized_start=1318 + _globals['_LOGGINGCONFIG_FILEDESTINATION']._serialized_end=1401 + _globals['_LOGGINGCONFIG_BIGQUERYDESTINATION']._serialized_start=1403 + _globals['_LOGGINGCONFIG_BIGQUERYDESTINATION']._serialized_end=1443 + _globals['_LOGGINGCONFIG_REDSHIFTDESTINATION']._serialized_start=1445 + _globals['_LOGGINGCONFIG_REDSHIFTDESTINATION']._serialized_end=1486 + _globals['_LOGGINGCONFIG_ATHENADESTINATION']._serialized_start=1488 + _globals['_LOGGINGCONFIG_ATHENADESTINATION']._serialized_end=1527 + _globals['_LOGGINGCONFIG_SNOWFLAKEDESTINATION']._serialized_start=1529 + _globals['_LOGGINGCONFIG_SNOWFLAKEDESTINATION']._serialized_end=1571 + _globals['_LOGGINGCONFIG_CUSTOMDESTINATION']._serialized_start=1574 + _globals['_LOGGINGCONFIG_CUSTOMDESTINATION']._serialized_end=1727 + _globals['_LOGGINGCONFIG_CUSTOMDESTINATION_CONFIGENTRY']._serialized_start=1682 + _globals['_LOGGINGCONFIG_CUSTOMDESTINATION_CONFIGENTRY']._serialized_end=1727 + _globals['_LOGGINGCONFIG_COUCHBASECOLUMNARDESTINATION']._serialized_start=1729 + _globals['_LOGGINGCONFIG_COUCHBASECOLUMNARDESTINATION']._serialized_end=1812 + _globals['_FEATURESERVICELIST']._serialized_start=1829 + _globals['_FEATURESERVICELIST']._serialized_end=1902 # @@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 a3e6158ad91..e05e302e5fd 100644 --- a/sdk/python/feast/protos/feast/core/FeatureService_pb2.pyi +++ b/sdk/python/feast/protos/feast/core/FeatureService_pb2.pyi @@ -66,6 +66,7 @@ class FeatureServiceSpec(google.protobuf.message.Message): OWNER_FIELD_NUMBER: builtins.int LOGGING_CONFIG_FIELD_NUMBER: builtins.int PRECOMPUTE_ONLINE_FIELD_NUMBER: builtins.int + PRECOMPUTE_OMIT_NAMES_FIELD_NUMBER: builtins.int name: builtins.str """Name of the Feature Service. Must be unique. Not updated.""" project: builtins.str @@ -89,6 +90,10 @@ class FeatureServiceSpec(google.protobuf.message.Message): """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. """ + precompute_omit_names: builtins.bool + """When true, pre-computed vectors store only values and a schema fingerprint, + not the repeated feature name list. + """ def __init__( self, *, @@ -100,9 +105,10 @@ class FeatureServiceSpec(google.protobuf.message.Message): owner: builtins.str = ..., logging_config: global___LoggingConfig | None = ..., precompute_online: builtins.bool = ..., + precompute_omit_names: 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", "precompute_online", b"precompute_online", "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_omit_names", b"precompute_omit_names", "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 index f86f32238cb..5873a4c950c 100644 --- a/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.py +++ b/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.py @@ -16,7 +16,7 @@ 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') +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\"\xde\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\x12\x1a\n\x12schema_fingerprint\x18\x05 \x01(\t\"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) @@ -25,7 +25,7 @@ _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 + _globals['_PRECOMPUTEDFEATUREVECTOR']._serialized_end=338 + _globals['_FEATUREVIEWTIMESTAMP']._serialized_start=340 + _globals['_FEATUREVIEWTIMESTAMP']._serialized_end=442 # @@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 index 919f1b59973..9d7a6ffe517 100644 --- a/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.pyi +++ b/sdk/python/feast/protos/feast/core/PrecomputedFeatureVector_pb2.pyi @@ -29,6 +29,7 @@ class PrecomputedFeatureVector(google.protobuf.message.Message): VALUES_FIELD_NUMBER: builtins.int FV_TIMESTAMPS_FIELD_NUMBER: builtins.int PRECOMPUTED_AT_FIELD_NUMBER: builtins.int + SCHEMA_FINGERPRINT_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".""" @@ -41,6 +42,10 @@ class PrecomputedFeatureVector(google.protobuf.message.Message): @property def precomputed_at(self) -> google.protobuf.timestamp_pb2.Timestamp: """Wall-clock time when this vector was assembled.""" + schema_fingerprint: builtins.str + """Fingerprint of the FeatureService's ordered "__" names. Lets a + vector omit feature_names and still be validated against the service. + """ def __init__( self, *, @@ -48,9 +53,10 @@ class PrecomputedFeatureVector(google.protobuf.message.Message): 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 = ..., + schema_fingerprint: builtins.str = ..., ) -> 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: ... + def ClearField(self, field_name: typing_extensions.Literal["feature_names", b"feature_names", "fv_timestamps", b"fv_timestamps", "precomputed_at", b"precomputed_at", "schema_fingerprint", b"schema_fingerprint", "values", b"values"]) -> None: ... global___PrecomputedFeatureVector = PrecomputedFeatureVector diff --git a/sdk/python/tests/unit/test_precomputed_feature_vectors.py b/sdk/python/tests/unit/test_precomputed_feature_vectors.py index ec7b4721ef5..eaade6eed9e 100644 --- a/sdk/python/tests/unit/test_precomputed_feature_vectors.py +++ b/sdk/python/tests/unit/test_precomputed_feature_vectors.py @@ -270,7 +270,14 @@ def test_validate_error_message_includes_service_and_odfv_name(self): class TestSchemaMismatchFallback: - def _fast_path(self, blobs, expected_names, grouped_refs=None, num_rows=None): + def _fast_path( + self, + blobs, + expected_names, + grouped_refs=None, + num_rows=None, + full_feature_names=True, + ): from feast.infra.online_stores.online_store import OnlineStore resolved_num_rows = num_rows or len(blobs) @@ -279,7 +286,7 @@ def _fast_path(self, blobs, expected_names, grouped_refs=None, num_rows=None): blobs=blobs, expected_feature_names=expected_names, online_features_response=response, - full_feature_names=True, + full_feature_names=full_feature_names, num_rows=resolved_num_rows, entity_row_indices=tuple([i] for i in range(len(blobs))), grouped_refs=grouped_refs or [], @@ -310,6 +317,86 @@ def test_matching_schema_returns_true(self): assert len(response.results) == 1 assert response.results[0].values[0].float_val == pytest.approx(42.0) + def test_bare_names_match_when_full_feature_names_is_false(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()], ["f1", "f2"], full_feature_names=False + ) + assert ok is True + assert list(response.metadata.feature_names.val) == ["f1", "f2"] + assert response.results[1].values[0].float_val == pytest.approx(2.0) + + def test_bare_names_reorder_when_full_feature_names_is_false(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()], ["f2", "f1"], full_feature_names=False + ) + 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 _nameless_service(self): + service = MagicMock() + service.precompute_feature_names.return_value = ["fv1__f1", "fv1__f2"] + service.precompute_schema_fingerprint.return_value = "abc123" + return service + + def test_nameless_vector_matches_on_fingerprint(self): + from feast.infra.online_stores.online_store import OnlineStore + + vector = _make_vector( + feature_names=[], + values=[ValueProto(float_val=1.0), ValueProto(float_val=2.0)], + ) + vector.schema_fingerprint = "abc123" + response = GetOnlineFeaturesResponse(results=[]) + ok = OnlineStore._try_precomputed_fast_path( + blobs=[vector.SerializeToString()], + expected_feature_names=["f2", "f1"], + online_features_response=response, + full_feature_names=False, + num_rows=1, + entity_row_indices=([0],), + grouped_refs=[], + registry=MagicMock(), + project="test", + feature_service=self._nameless_service(), + ) + 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_nameless_vector_with_stale_fingerprint_returns_false(self): + from feast.infra.online_stores.online_store import OnlineStore + + vector = _make_vector(feature_names=[], values=[ValueProto(float_val=1.0)]) + vector.schema_fingerprint = "stale" + ok = OnlineStore._try_precomputed_fast_path( + blobs=[vector.SerializeToString()], + expected_feature_names=["f1"], + online_features_response=GetOnlineFeaturesResponse(results=[]), + full_feature_names=False, + num_rows=1, + entity_row_indices=([0],), + grouped_refs=[], + registry=MagicMock(), + project="test", + feature_service=self._nameless_service(), + ) + assert ok is False + + def test_nameless_vector_without_service_returns_false(self): + vector = _make_vector(feature_names=[], values=[ValueProto(float_val=1.0)]) + vector.schema_fingerprint = "abc123" + ok, _ = self._fast_path([vector.SerializeToString()], ["fv1__f1"]) + assert ok is False + def test_extra_feature_in_expected_returns_false(self): vector = _make_vector( feature_names=["fv1__f1"], @@ -1008,7 +1095,18 @@ def _make_mock_store(self, services): store._precompute_for_push = FeatureStore._precompute_for_push.__get__(store) return store + def _make_pushed_fv(self): + from feast.field import Field + from feast.types import Int64 + + fv = MagicMock() + fv.join_keys = ["driver_id"] + fv.entity_columns = [Field(name="driver_id", dtype=Int64)] + return fv + def test_triggers_for_matching_fv(self): + import pandas as pd + proj = MagicMock() proj.name = "pushed_fv" svc = MagicMock() @@ -1017,8 +1115,33 @@ def test_triggers_for_matching_fv(self): 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") + store.registry.get_any_feature_view.return_value = self._make_pushed_fv() + df = pd.DataFrame( + {"driver_id": [1001, 1002, 1001], "conv_rate": [0.1, 0.2, 0.3]} + ) + store._precompute_for_push("pushed_fv", df) + + store.precompute_feature_service.assert_called_once() + args, kwargs = store.precompute_feature_service.call_args + assert args == ("svc",) + entity_keys = kwargs["entity_keys"] + assert [ek.entity_values[0].int64_val for ek in entity_keys] == [1001, 1002] + assert all(list(ek.join_keys) == ["driver_id"] for ek in entity_keys) + + def test_skips_when_pushed_rows_have_no_entity_keys(self): + import pandas as pd + + 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.registry.get_any_feature_view.return_value = self._make_pushed_fv() + store._precompute_for_push("pushed_fv", pd.DataFrame({"conv_rate": [0.1]})) + store.precompute_feature_service.assert_not_called() def test_skips_non_matching_fv(self): proj = MagicMock() @@ -1065,3 +1188,46 @@ def test_handles_registry_exception_gracefully(self): from feast.feature_store import FeatureStore FeatureStore._precompute_for_push(store, "fv", MagicMock()) + + +class TestPrecomputeOmitNames: + def _service(self, omit): + fv = _make_fv( + "fv1", [Field(name="f1", dtype=Float32), Field(name="f2", dtype=Float32)] + ) + return FeatureService( + name="svc", + features=[fv], + precompute_online=True, + precompute_omit_names=omit, + ) + + def test_flag_round_trips_through_proto(self): + service = self._service(True) + assert FeatureService.from_proto(service.to_proto()).precompute_omit_names + assert not FeatureService.from_proto( + self._service(False).to_proto() + ).precompute_omit_names + + def test_fingerprint_follows_ordered_names(self): + service = self._service(True) + assert service.precompute_feature_names() == ["fv1__f1", "fv1__f2"] + reordered = FeatureService( + name="svc2", + features=[ + _make_fv( + "fv1", + [Field(name="f2", dtype=Float32), Field(name="f1", dtype=Float32)], + ) + ], + precompute_online=True, + precompute_omit_names=True, + ) + assert ( + service.precompute_schema_fingerprint() + != reordered.precompute_schema_fingerprint() + ) + assert ( + service.precompute_schema_fingerprint() + == self._service(True).precompute_schema_fingerprint() + )