diff --git a/sdk/python/feast/stream_feature_view.py b/sdk/python/feast/stream_feature_view.py index 00aa239266b..760f2848699 100644 --- a/sdk/python/feast/stream_feature_view.py +++ b/sdk/python/feast/stream_feature_view.py @@ -311,7 +311,7 @@ def to_proto(self): name=self.name, entities=self.entities, entity_columns=[field.to_proto() for field in self.entity_columns], - features=[field.to_proto() for field in self.schema], + features=[field.to_proto() for field in self.features], user_defined_function=udf_proto, feature_transformation=feature_transformation, description=self.description, diff --git a/sdk/python/tests/unit/test_stream_feature_view.py b/sdk/python/tests/unit/test_stream_feature_view.py index 57eac326c10..19c79383014 100644 --- a/sdk/python/tests/unit/test_stream_feature_view.py +++ b/sdk/python/tests/unit/test_stream_feature_view.py @@ -13,7 +13,7 @@ StreamFeatureView as StreamFeatureViewProto, ) from feast.stream_feature_view import StreamFeatureView, stream_feature_view -from feast.types import Float32 +from feast.types import Float32, Int64 from feast.utils import _utc_now, make_tzaware @@ -112,6 +112,41 @@ def test_stream_feature_view_serialization(): ) +def test_stream_feature_view_serialization_excludes_entity_columns_from_features(): + entity = Entity(name="driver_entity", join_keys=["driver_id"]) + stream_source = KafkaSource( + name="kafka", + timestamp_field="event_timestamp", + kafka_bootstrap_servers="", + message_format=AvroFormat(""), + topic="topic", + batch_source=FileSource(path="some path"), + ) + + sfv = StreamFeatureView( + name="test kafka stream feature view", + entities=[entity], + ttl=timedelta(days=30), + schema=[ + Field(name="driver_id", dtype=Int64), + Field(name="dummy_field", dtype=Float32), + ], + timestamp_field="event_timestamp", + source=stream_source, + ) + assert [f.name for f in sfv.features] == ["dummy_field"] + assert [f.name for f in sfv.entity_columns] == ["driver_id"] + + sfv_proto = sfv.to_proto() + assert [f.name for f in sfv_proto.spec.features] == ["dummy_field"] + assert [f.name for f in sfv_proto.spec.entity_columns] == ["driver_id"] + + new_sfv = StreamFeatureView.from_proto(sfv_proto=sfv_proto) + assert new_sfv.features == sfv.features + assert new_sfv.entity_columns == sfv.entity_columns + assert new_sfv == sfv + + def test_stream_feature_view_udfs(): entity = Entity(name="driver_entity", join_keys=["test_key"]) stream_source = KafkaSource(