diff --git a/sdk/python/feast/infra/online_stores/sqlite.py b/sdk/python/feast/infra/online_stores/sqlite.py index d56c430a996..000e3d7bc5a 100644 --- a/sdk/python/feast/infra/online_stores/sqlite.py +++ b/sdk/python/feast/infra/online_stores/sqlite.py @@ -60,6 +60,7 @@ from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto from feast.repo_config import FeastConfigBaseModel, RepoConfig +from feast.stream_feature_view import StreamFeatureView from feast.type_map import feast_value_type_to_python_type from feast.types import FEAST_VECTOR_TYPES, PrimitiveFeastType from feast.utils import ( @@ -465,20 +466,23 @@ def plan( config.online_store, "enable_openai_compatible_store", False ) + # FeatureView.from_proto() is @typechecked and only accepts a + # FeatureViewProto, so it can't be applied to stream_feature_views + # (StreamFeatureViewProto) too -- each list needs its matching class. + views = [ + FeatureView.from_proto(view) + for view in desired_registry_proto.feature_views + ] + [ + StreamFeatureView.from_proto(view) + for view in desired_registry_proto.stream_feature_views + ] infra_objects: List[InfraObject] = [ SqliteTable( path=self._get_db_path(config), - name=_table_id( - project, - FeatureView.from_proto(view), - versioning, - ), + name=_table_id(project, view, versioning), include_value_num=include_value_num, ) - for view in [ - *desired_registry_proto.feature_views, - *desired_registry_proto.stream_feature_views, - ] + for view in views ] for lv_proto in desired_registry_proto.label_views: diff --git a/sdk/python/tests/unit/infra/online_store/test_sqlite_plan.py b/sdk/python/tests/unit/infra/online_store/test_sqlite_plan.py new file mode 100644 index 00000000000..b01a0650e52 --- /dev/null +++ b/sdk/python/tests/unit/infra/online_store/test_sqlite_plan.py @@ -0,0 +1,88 @@ +from datetime import timedelta + +from feast.data_format import AvroFormat +from feast.data_source import KafkaSource +from feast.feature_view import FeatureView +from feast.field import Field +from feast.infra.offline_stores.dask import DaskOfflineStoreConfig +from feast.infra.offline_stores.file_source import FileSource +from feast.infra.online_stores.sqlite import SqliteOnlineStore, SqliteOnlineStoreConfig +from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto +from feast.repo_config import RepoConfig +from feast.stream_feature_view import StreamFeatureView +from feast.types import String + + +def _repo_config() -> RepoConfig: + return RepoConfig( + registry="/tmp/unused_registry.db", + project="test_project", + provider="local", + online_store=SqliteOnlineStoreConfig(), + offline_store=DaskOfflineStoreConfig(), + entity_key_serialization_version=3, + ) + + +def _feature_view(name: str) -> FeatureView: + return FeatureView( + name=name, + entities=[], + schema=[Field(name="value", dtype=String)], + ttl=timedelta(days=1), + online=True, + source=FileSource(path="dummy.parquet", timestamp_field="event_timestamp"), + ) + + +def _stream_feature_view(name: str) -> StreamFeatureView: + return StreamFeatureView( + name=name, + entities=[], + schema=[Field(name="value", dtype=String)], + source=KafkaSource( + name="dummy_kafka", + timestamp_field="event_timestamp", + message_format=AvroFormat(""), + kafka_bootstrap_servers="localhost:9092", + topic="dummy_topic", + batch_source=FileSource( + path="dummy.parquet", timestamp_field="event_timestamp" + ), + ), + ) + + +class TestSqliteOnlineStorePlanWithStreamFeatureViews: + """Regression test for a typeguard.TypeCheckError previously raised by + plan() when the registry contains a StreamFeatureView: FeatureView.from_proto() + was applied uniformly to both feature_views and stream_feature_views, but + FeatureView is @typechecked and a StreamFeatureView proto is not a + FeatureView proto.""" + + def test_plan_succeeds_with_only_stream_feature_views(self): + config = _repo_config() + registry_proto = RegistryProto() + registry_proto.stream_feature_views.append( + _stream_feature_view("driver_dropoffs_stream").to_proto() + ) + + infra_objects = SqliteOnlineStore().plan(config, registry_proto) + + assert len(infra_objects) == 1 + assert infra_objects[0].name == "test_project_driver_dropoffs_stream" + + def test_plan_succeeds_with_batch_and_stream_feature_views_together(self): + config = _repo_config() + registry_proto = RegistryProto() + registry_proto.feature_views.append(_feature_view("batch_view").to_proto()) + registry_proto.stream_feature_views.append( + _stream_feature_view("driver_dropoffs_stream").to_proto() + ) + + infra_objects = SqliteOnlineStore().plan(config, registry_proto) + + assert sorted(o.name for o in infra_objects) == [ + "test_project_batch_view", + "test_project_driver_dropoffs_stream", + ]