Skip to content

Commit 9420fac

Browse files
chandlerokclaudeshuchu
authored
feat: Optional nameless pre-computed vectors with a schema fingerprint (#7077)
* feat: Refresh pre-computed vectors for pushed entities only precompute_feature_service accepts entity_keys and skips the keyspace scan when they are given. The push hook now passes the distinct entity keys from the pushed DataFrame instead of rebuilding every entity in the store. Signed-off-by: Chandler King <[email protected]> Co-Authored-By: Claude Fable 5.1 <[email protected]> * fix: Match pre-computed vector names when full_feature_names is False Stored vectors carry "<fv>__<feature>" names while the default request expects bare feature names, so the fast path always reported a schema mismatch unless full_feature_names=True. Compare on the caller's naming and derive the per-feature view name from the stored names for TTL checks. Signed-off-by: Chandler King <[email protected]> Co-Authored-By: Claude Fable 5.1 <[email protected]> * feat: Optional nameless pre-computed vectors with a schema fingerprint A pre-computed vector stores the full repeated feature name list next to its values. For a wide service that is most of the blob: 90 features came to 4.8 KB per entity, 3.8 KB of it names that are identical for every entity. FeatureService gains precompute_omit_names. Vectors written for such a service carry only values, per-view timestamps, and a fingerprint of the service's ordered names. The fast path validates a nameless vector by fingerprint and takes its names from the service. Vectors with names keep working as before. With the flag on the same blob is 834 bytes, and a 500-entity read moves 414 KB instead of 2.4 MB. Generated code is from grpcio-tools 1.62.3 (protobuf 4.25) and mypy-protobuf 3.3.0 to match the checked-in files. Signed-off-by: Chandler King <[email protected]> Co-Authored-By: Claude Fable 5.1 <[email protected]> * chore: Retrigger CI Signed-off-by: Chandler King <[email protected]> --------- Signed-off-by: Chandler King <[email protected]> Co-authored-by: Claude Fable 5.1 <[email protected]> Co-authored-by: Shuchu Han <[email protected]>
1 parent 1396c88 commit 9420fac

10 files changed

Lines changed: 380 additions & 89 deletions

‎protos/feast/core/FeatureService.proto‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,10 @@ message FeatureServiceSpec {
4242
// When true, a pre-computed feature vector is maintained per entity for this
4343
// service, enabling single-read online retrieval instead of per-feature-view reads.
4444
bool precompute_online = 8;
45+
46+
// When true, pre-computed vectors store only values and a schema fingerprint,
47+
// not the repeated feature name list.
48+
bool precompute_omit_names = 9;
4549
}
4650

4751

‎protos/feast/core/PrecomputedFeatureVector.proto‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,10 @@ message PrecomputedFeatureVector {
2222

2323
// Wall-clock time when this vector was assembled.
2424
google.protobuf.Timestamp precomputed_at = 4;
25+
26+
// Fingerprint of the FeatureService's ordered "<fv>__<feature>" names. Lets a
27+
// vector omit feature_names and still be validated against the service.
28+
string schema_fingerprint = 5;
2529
}
2630

2731
// Event timestamp associated with a specific feature view within the vector.

‎sdk/python/feast/feature_service.py‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import copy
2+
import hashlib
23
from datetime import datetime
34
from typing import TYPE_CHECKING, Dict, List, Optional, Union
45

@@ -61,6 +62,7 @@ class FeatureService:
6162
logging_config: Optional[LoggingConfig] = None
6263

6364
precompute_online: bool = False
65+
precompute_omit_names: bool = False
6466

6567
def __init__(
6668
self,
@@ -72,6 +74,7 @@ def __init__(
7274
owner: str = "",
7375
logging_config: Optional[LoggingConfig] = None,
7476
precompute_online: bool = False,
77+
precompute_omit_names: bool = False,
7578
):
7679
"""
7780
Creates a FeatureService object.
@@ -108,6 +111,7 @@ def __init__(
108111
self.last_updated_timestamp = None
109112
self.logging_config = logging_config
110113
self.precompute_online = precompute_online
114+
self.precompute_omit_names = precompute_omit_names
111115
for feature_grouping in self._features:
112116
if isinstance(feature_grouping, str):
113117
# No registry at construction time; resolved in resolve_pending_refs.
@@ -360,6 +364,7 @@ def prepare_for_apply(
360364
owner=self.owner,
361365
logging_config=self.logging_config,
362366
precompute_online=self.precompute_online,
367+
precompute_omit_names=self.precompute_omit_names,
363368
)
364369
prepared.created_timestamp = self.created_timestamp
365370
prepared.last_updated_timestamp = self.last_updated_timestamp
@@ -394,6 +399,20 @@ def __str__(self):
394399
def __hash__(self):
395400
return hash(self.name)
396401

402+
def precompute_feature_names(self) -> List[str]:
403+
"""``<fv>__<feature>`` names in the order pre-computed vectors store values."""
404+
return [
405+
f"{projection.name_to_use()}__{feature.name}"
406+
for projection in self.feature_view_projections
407+
for feature in projection.features
408+
]
409+
410+
def precompute_schema_fingerprint(self) -> str:
411+
"""Stable digest of :meth:`precompute_feature_names` for nameless vectors."""
412+
return hashlib.sha256(
413+
"\n".join(self.precompute_feature_names()).encode("utf-8")
414+
).hexdigest()[:16]
415+
397416
def __eq__(self, other):
398417
if not isinstance(other, FeatureService):
399418
return False
@@ -404,6 +423,7 @@ def __eq__(self, other):
404423
or self.tags != other.tags
405424
or self.owner != other.owner
406425
or self.precompute_online != other.precompute_online
426+
or self.precompute_omit_names != other.precompute_omit_names
407427
):
408428
return False
409429

@@ -432,6 +452,7 @@ def from_proto(cls, feature_service_proto: FeatureServiceProto):
432452
feature_service_proto.spec.logging_config
433453
),
434454
precompute_online=feature_service_proto.spec.precompute_online,
455+
precompute_omit_names=feature_service_proto.spec.precompute_omit_names,
435456
)
436457
fs.feature_view_projections.extend(
437458
[
@@ -484,6 +505,7 @@ def to_proto(self) -> FeatureServiceProto:
484505
if self.logging_config
485506
else None,
486507
precompute_online=self.precompute_online,
508+
precompute_omit_names=self.precompute_omit_names,
487509
)
488510

489511
return FeatureServiceProto(spec=spec, meta=meta)

‎sdk/python/feast/feature_store.py‎

Lines changed: 89 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,7 @@
115115
_get_feature_view_vector_field_metadata,
116116
_utc_now,
117117
)
118+
from feast.value_type import ValueType
118119
from feast.vector_store_utils import feature_view_to_vs_id
119120
from feast.version_utils import parse_version
120121

@@ -3193,6 +3194,7 @@ def precompute_feature_service(
31933194
self,
31943195
feature_service_name: Optional[str] = None,
31953196
batch_size: int = 1000,
3197+
entity_keys: Optional[List[EntityKey]] = None,
31963198
) -> int:
31973199
"""Pre-compute feature vectors for one or all FeatureServices.
31983200
@@ -3203,6 +3205,9 @@ def precompute_feature_service(
32033205
32043206
Works with **all** online store backends (Redis, DynamoDB, PostgreSQL, etc.).
32053207
3208+
When ``entity_keys`` is given, only those entities are rebuilt and the store
3209+
is not scanned, so a push can refresh exactly the rows it wrote.
3210+
32063211
Returns the total number of entity vectors written.
32073212
"""
32083213
from feast.protos.feast.core.PrecomputedFeatureVector_pb2 import (
@@ -3250,53 +3255,56 @@ def precompute_feature_service(
32503255
# For stores that support native scanning (like Redis), we try the
32513256
# native scan; for others, we use the materialized entity list.
32523257
entity_key_protos: List[EntityKey] = []
3253-
try:
3254-
from feast.infra.online_stores.redis import RedisOnlineStore
3258+
if entity_keys is not None:
3259+
entity_key_protos = list(entity_keys)
3260+
else:
3261+
try:
3262+
from feast.infra.online_stores.redis import RedisOnlineStore
32553263

3256-
if isinstance(online_store, RedisOnlineStore):
3257-
from feast.infra.key_encoding_utils import (
3258-
deserialize_entity_key,
3259-
)
3260-
from feast.infra.online_stores.helpers import _redis_key_prefix
3261-
3262-
join_keys = (
3263-
list(primary_fv.join_keys)
3264-
if hasattr(primary_fv, "join_keys")
3265-
else []
3266-
) # type: ignore[union-attr]
3267-
client = online_store._get_client( # type: ignore[attr-defined]
3268-
self.config.online_store
3264+
if isinstance(online_store, RedisOnlineStore):
3265+
from feast.infra.key_encoding_utils import (
3266+
deserialize_entity_key,
3267+
)
3268+
from feast.infra.online_stores.helpers import _redis_key_prefix
3269+
3270+
join_keys = (
3271+
list(primary_fv.join_keys)
3272+
if hasattr(primary_fv, "join_keys")
3273+
else []
3274+
) # type: ignore[union-attr]
3275+
client = online_store._get_client( # type: ignore[attr-defined]
3276+
self.config.online_store
3277+
)
3278+
scan_prefix = _redis_key_prefix(join_keys)
3279+
project_bytes = self.config.project.encode("utf-8")
3280+
seen_keys: set = set()
3281+
raw_keys: List[bytes] = []
3282+
for key in client.scan_iter(
3283+
b"".join([scan_prefix, b"*", project_bytes])
3284+
):
3285+
if key not in seen_keys:
3286+
seen_keys.add(key)
3287+
raw_keys.append(key)
3288+
proj_len = len(project_bytes)
3289+
for rk in raw_keys:
3290+
try:
3291+
ek = deserialize_entity_key(
3292+
rk[:-proj_len],
3293+
self.config.entity_key_serialization_version,
3294+
)
3295+
if set(ek.join_keys) == set(join_keys):
3296+
entity_key_protos.append(ek)
3297+
except (ValueError, Exception):
3298+
continue
3299+
else:
3300+
raise NotImplementedError
3301+
except (ImportError, NotImplementedError):
3302+
_logger.warning(
3303+
"Entity scanning not supported for '%s' — "
3304+
"precompute requires a store that supports entity scanning",
3305+
type(online_store).__name__,
32693306
)
3270-
scan_prefix = _redis_key_prefix(join_keys)
3271-
project_bytes = self.config.project.encode("utf-8")
3272-
seen_keys: set = set()
3273-
raw_keys: List[bytes] = []
3274-
for key in client.scan_iter(
3275-
b"".join([scan_prefix, b"*", project_bytes])
3276-
):
3277-
if key not in seen_keys:
3278-
seen_keys.add(key)
3279-
raw_keys.append(key)
3280-
proj_len = len(project_bytes)
3281-
for rk in raw_keys:
3282-
try:
3283-
ek = deserialize_entity_key(
3284-
rk[:-proj_len],
3285-
self.config.entity_key_serialization_version,
3286-
)
3287-
if set(ek.join_keys) == set(join_keys):
3288-
entity_key_protos.append(ek)
3289-
except (ValueError, Exception):
3290-
continue
3291-
else:
3292-
raise NotImplementedError
3293-
except (ImportError, NotImplementedError):
3294-
_logger.warning(
3295-
"Entity scanning not supported for '%s' — "
3296-
"precompute requires a store that supports entity scanning",
3297-
type(online_store).__name__,
3298-
)
3299-
continue
3307+
continue
33003308

33013309
if not entity_key_protos:
33023310
continue
@@ -3348,10 +3356,13 @@ def precompute_feature_service(
33483356
now_ts.FromDatetime(datetime.now(tz=_timezone.utc))
33493357

33503358
vector = PrecomputedFeatureVector(
3351-
feature_names=feature_names,
3359+
feature_names=[]
3360+
if svc.precompute_omit_names
3361+
else feature_names,
33523362
values=values,
33533363
fv_timestamps=fv_timestamps,
33543364
precomputed_at=now_ts,
3365+
schema_fingerprint=svc.precompute_schema_fingerprint(),
33553366
)
33563367

33573368
online_store.write_precomputed_vector(
@@ -3409,9 +3420,16 @@ def _precompute_for_push(self, feature_view_name: str, df: "pd.DataFrame") -> No
34093420
if not affected:
34103421
return
34113422

3423+
fv = self.registry.get_any_feature_view(
3424+
feature_view_name, self.project, allow_cache=True
3425+
)
3426+
entity_keys = _entity_keys_from_dataframe(fv, df)
3427+
if not entity_keys:
3428+
return
3429+
34123430
for svc in affected:
34133431
try:
3414-
self.precompute_feature_service(svc.name)
3432+
self.precompute_feature_service(svc.name, entity_keys=entity_keys)
34153433
except Exception:
34163434
_logger.warning(
34173435
"Failed to precompute vectors for service '%s' after push",
@@ -5222,3 +5240,27 @@ def _validate_data_sources(
52225240
UserWarning,
52235241
stacklevel=2,
52245242
)
5243+
5244+
5245+
def _entity_keys_from_dataframe(feature_view: Any, df: pd.DataFrame) -> List[EntityKey]:
5246+
"""Distinct entity keys present in ``df`` for ``feature_view``'s join keys."""
5247+
from feast.type_map import python_values_to_proto_values
5248+
5249+
join_keys = list(getattr(feature_view, "join_keys", []) or [])
5250+
if not join_keys or any(k not in df.columns for k in join_keys):
5251+
return []
5252+
value_types = {
5253+
field.name: field.dtype.to_value_type()
5254+
for field in getattr(feature_view, "entity_columns", [])
5255+
}
5256+
unique = df[join_keys].drop_duplicates()
5257+
columns = {
5258+
k: python_values_to_proto_values(
5259+
unique[k].tolist(), value_types.get(k, ValueType.UNKNOWN)
5260+
)
5261+
for k in join_keys
5262+
}
5263+
return [
5264+
EntityKey(join_keys=join_keys, entity_values=[columns[k][i] for k in join_keys])
5265+
for i in range(len(unique))
5266+
]

‎sdk/python/feast/infra/online_stores/online_store.py‎

Lines changed: 47 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -241,6 +241,7 @@ def get_online_features(
241241
grouped_refs,
242242
registry,
243243
project,
244+
feature_service=features,
244245
)
245246
if not precomputed_ok:
246247
raise RuntimeError(
@@ -527,6 +528,7 @@ async def get_online_features_async(
527528
grouped_refs,
528529
registry,
529530
project,
531+
feature_service=features,
530532
)
531533
if not precomputed_ok:
532534
raise RuntimeError(
@@ -612,6 +614,7 @@ def _try_precomputed_fast_path(
612614
grouped_refs: List,
613615
registry: BaseRegistry,
614616
project: str,
617+
feature_service: Optional[Any] = None,
615618
) -> bool:
616619
"""Build the response from pre-computed vectors.
617620
@@ -662,27 +665,65 @@ def _try_precomputed_fast_path(
662665
reorder_map: Optional[List[int]] = None
663666
vectors: List[PrecomputedFeatureVector] = []
664667

668+
# A nameless vector is validated by fingerprint and takes its names, in
669+
# stored order, from the FeatureService it was built for.
670+
service_names: Optional[List[str]] = None
671+
service_fingerprint: Optional[str] = None
672+
if feature_service is not None and hasattr(
673+
feature_service, "precompute_schema_fingerprint"
674+
):
675+
service_names = feature_service.precompute_feature_names()
676+
service_fingerprint = feature_service.precompute_schema_fingerprint()
677+
665678
for blob in blobs:
666679
vec = PrecomputedFeatureVector()
667680
vec.ParseFromString(blob) # type: ignore[arg-type]
668681
vectors.append(vec)
669682

683+
stored_names = list(vec.feature_names)
684+
if not stored_names:
685+
if (
686+
service_names is None
687+
or vec.schema_fingerprint != service_fingerprint
688+
):
689+
logger.warning(
690+
"Pre-computed vector fingerprint mismatch at index %d: stored=%s, expected=%s",
691+
len(vectors) - 1,
692+
vec.schema_fingerprint,
693+
service_fingerprint,
694+
)
695+
return False
696+
stored_names = service_names
697+
670698
if first_stored_names is None:
671-
first_stored_names = list(vec.feature_names)
672-
if set(first_stored_names) != expected_set:
699+
first_stored_names = stored_names
700+
# Vectors always store "<fv>__<feature>"; the caller's expected
701+
# names are bare unless full_feature_names was requested.
702+
comparable_names = (
703+
first_stored_names
704+
if full_feature_names
705+
else [n.split("__", 1)[-1] for n in first_stored_names]
706+
)
707+
if set(comparable_names) != expected_set:
673708
logger.warning(
674709
"Pre-computed vector schema mismatch: stored=%s, expected=%s",
675710
first_stored_names,
676711
expected_feature_names,
677712
)
678713
return False
679-
if first_stored_names != expected_feature_names:
680-
name_to_idx = {n: i for i, n in enumerate(first_stored_names)}
714+
if comparable_names != expected_feature_names:
715+
name_to_idx = {n: i for i, n in enumerate(comparable_names)}
681716
reorder_map = [name_to_idx[n] for n in expected_feature_names]
717+
feat_fv_names = [
718+
first_stored_names[reorder_map[i] if reorder_map else i].split(
719+
"__", 1
720+
)[0]
721+
for i in range(n_features)
722+
]
682723
else:
683724
if (
684-
len(vec.feature_names) != n_features
685-
or list(vec.feature_names) != first_stored_names
725+
len(stored_names) != n_features
726+
or stored_names != first_stored_names
686727
):
687728
logger.warning(
688729
"Pre-computed vector schema varies across entities at index %d",

0 commit comments

Comments
 (0)