feat: Optional nameless pre-computed vectors with a schema fingerprint

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;
 }
 
 

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 "<fv>__<feature>" 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.

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]:
+        """``<fv>__<feature>`` 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)

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))
+    ]

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 "<fv>__<feature>"; 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",