Visitar URL original
fix: use matching proto class per feature view list in SqliteOnlineSt… · feast-dev/feast@02e0617 · GitHub
Skip to content

Commit 02e0617

Browse files
author
Carlos Sánchez
committed
fix: use matching proto class per feature view list in SqliteOnlineStore.plan()
FeatureView.from_proto() was applied uniformly to both feature_views and stream_feature_views. FeatureView is @TypeChecked and its from_proto() parameter is annotated as FeatureViewProto, so passing a StreamFeatureViewProto raises typeguard.TypeCheckError at runtime. Use StreamFeatureView.from_proto() for the stream_feature_views list instead, matching the actual proto type of each entry. Fixes #6658
1 parent 104ad10 commit 02e0617

2 files changed

Lines changed: 101 additions & 9 deletions

File tree

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

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@
6060
from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
6161
from feast.protos.feast.types.Value_pb2 import Value as ValueProto
6262
from feast.repo_config import FeastConfigBaseModel, RepoConfig
63+
from feast.stream_feature_view import StreamFeatureView
6364
from feast.type_map import feast_value_type_to_python_type
6465
from feast.types import FEAST_VECTOR_TYPES, PrimitiveFeastType
6566
from feast.utils import (
@@ -465,20 +466,23 @@ def plan(
465466
config.online_store, "enable_openai_compatible_store", False
466467
)
467468

469+
# FeatureView.from_proto() is @typechecked and only accepts a
470+
# FeatureViewProto, so it can't be applied to stream_feature_views
471+
# (StreamFeatureViewProto) too -- each list needs its matching class.
472+
views = [
473+
FeatureView.from_proto(view)
474+
for view in desired_registry_proto.feature_views
475+
] + [
476+
StreamFeatureView.from_proto(view)
477+
for view in desired_registry_proto.stream_feature_views
478+
]
468479
infra_objects: List[InfraObject] = [
469480
SqliteTable(
470481
path=self._get_db_path(config),
471-
name=_table_id(
472-
project,
473-
FeatureView.from_proto(view),
474-
versioning,
475-
),
482+
name=_table_id(project, view, versioning),
476483
include_value_num=include_value_num,
477484
)
478-
for view in [
479-
*desired_registry_proto.feature_views,
480-
*desired_registry_proto.stream_feature_views,
481-
]
485+
for view in views
482486
]
483487

484488
for lv_proto in desired_registry_proto.label_views:

‎sdk/python/tests/unit/infra/online_store/test_sqlite_plan.py‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
from datetime import timedelta
2+
3+
from feast.data_format import AvroFormat
4+
from feast.data_source import KafkaSource
5+
from feast.feature_view import FeatureView
6+
from feast.field import Field
7+
from feast.infra.offline_stores.dask import DaskOfflineStoreConfig
8+
from feast.infra.offline_stores.file_source import FileSource
9+
from feast.infra.online_stores.sqlite import SqliteOnlineStore, SqliteOnlineStoreConfig
10+
from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto
11+
from feast.repo_config import RepoConfig
12+
from feast.stream_feature_view import StreamFeatureView
13+
from feast.types import String
14+
15+
16+
def _repo_config() -> RepoConfig:
17+
return RepoConfig(
18+
registry="/tmp/unused_registry.db",
19+
project="test_project",
20+
provider="local",
21+
online_store=SqliteOnlineStoreConfig(),
22+
offline_store=DaskOfflineStoreConfig(),
23+
entity_key_serialization_version=3,
24+
)
25+
26+
27+
def _feature_view(name: str) -> FeatureView:
28+
return FeatureView(
29+
name=name,
30+
entities=[],
31+
schema=[Field(name="value", dtype=String)],
32+
ttl=timedelta(days=1),
33+
online=True,
34+
source=FileSource(path="dummy.parquet", timestamp_field="event_timestamp"),
35+
)
36+
37+
38+
def _stream_feature_view(name: str) -> StreamFeatureView:
39+
return StreamFeatureView(
40+
name=name,
41+
entities=[],
42+
schema=[Field(name="value", dtype=String)],
43+
source=KafkaSource(
44+
name="dummy_kafka",
45+
timestamp_field="event_timestamp",
46+
message_format=AvroFormat(""),
47+
kafka_bootstrap_servers="localhost:9092",
48+
topic="dummy_topic",
49+
batch_source=FileSource(
50+
path="dummy.parquet", timestamp_field="event_timestamp"
51+
),
52+
),
53+
)
54+
55+
56+
class TestSqliteOnlineStorePlanWithStreamFeatureViews:
57+
"""Regression test for a typeguard.TypeCheckError previously raised by
58+
plan() when the registry contains a StreamFeatureView: FeatureView.from_proto()
59+
was applied uniformly to both feature_views and stream_feature_views, but
60+
FeatureView is @typechecked and a StreamFeatureView proto is not a
61+
FeatureView proto."""
62+
63+
def test_plan_succeeds_with_only_stream_feature_views(self):
64+
config = _repo_config()
65+
registry_proto = RegistryProto()
66+
registry_proto.stream_feature_views.append(
67+
_stream_feature_view("driver_dropoffs_stream").to_proto()
68+
)
69+
70+
infra_objects = SqliteOnlineStore().plan(config, registry_proto)
71+
72+
assert len(infra_objects) == 1
73+
assert infra_objects[0].name == "test_project_driver_dropoffs_stream"
74+
75+
def test_plan_succeeds_with_batch_and_stream_feature_views_together(self):
76+
config = _repo_config()
77+
registry_proto = RegistryProto()
78+
registry_proto.feature_views.append(_feature_view("batch_view").to_proto())
79+
registry_proto.stream_feature_views.append(
80+
_stream_feature_view("driver_dropoffs_stream").to_proto()
81+
)
82+
83+
infra_objects = SqliteOnlineStore().plan(config, registry_proto)
84+
85+
assert sorted(o.name for o in infra_objects) == [
86+
"test_project_batch_view",
87+
"test_project_driver_dropoffs_stream",
88+
]

0 commit comments

Comments
 (0)