diff --git a/sdk/python/feast/stream_feature_view.py b/sdk/python/feast/stream_feature_view.py index 00aa239266b..89c7a2efbdd 100644 --- a/sdk/python/feast/stream_feature_view.py +++ b/sdk/python/feast/stream_feature_view.py @@ -253,17 +253,18 @@ def __eq__(self, other): if not super().__eq__(other): return False - if not self.udf: - return not other.udf - if not other.udf: + if bool(self.udf) != bool(other.udf): + return False + if self.udf and self.udf.__code__.co_code != other.udf.__code__.co_code: return False if ( - self.mode != other.mode + mode_to_string(self.mode) != mode_to_string(other.mode) or self.timestamp_field != other.timestamp_field - or self.udf.__code__.co_code != other.udf.__code__.co_code - or self.udf_string != other.udf_string + or (self.udf_string or "") != (other.udf_string or "") or self.aggregations != other.aggregations + or self.enable_tiling != other.enable_tiling + or self.tiling_hop_size != other.tiling_hop_size ): return False diff --git a/sdk/python/tests/unit/test_stream_feature_view.py b/sdk/python/tests/unit/test_stream_feature_view.py index 57eac326c10..c585719096b 100644 --- a/sdk/python/tests/unit/test_stream_feature_view.py +++ b/sdk/python/tests/unit/test_stream_feature_view.py @@ -345,3 +345,93 @@ def test_stream_feature_view_org_field(): sfv_other_org = StreamFeatureView(name="sfv-with-org", org="search", **common) assert sfv_with_org != sfv_other_org + + +@pytest.mark.parametrize( + "changes", + [ + { + "aggregations": [ + Aggregation( + column="dummy_field", + function="max", + time_window=timedelta(days=7), + ) + ] + }, + {"mode": "pandas"}, + {"timestamp_field": "other_timestamp"}, + {"enable_tiling": True}, + {"enable_tiling": True, "tiling_hop_size": timedelta(minutes=10)}, + ], +) +def test_stream_feature_view_equality_without_udf(changes): + entity = Entity(name="driver_entity", join_keys=["test_key"]) + stream_source = KafkaSource( + name="kafka", + timestamp_field="event_timestamp", + kafka_bootstrap_servers="", + message_format=AvroFormat(""), + topic="topic", + batch_source=FileSource(path="some path"), + ) + + def make_sfv(**kwargs): + args = dict( + name="test kafka stream feature view", + entities=[entity], + ttl=timedelta(days=30), + schema=[Field(name="dummy_field", dtype=Float32)], + source=stream_source, + timestamp_field="event_timestamp", + aggregations=[ + Aggregation( + column="dummy_field", + function="sum", + time_window=timedelta(hours=1), + ) + ], + ) + args.update(kwargs) + return StreamFeatureView(**args) + + sfv = make_sfv() + assert sfv.udf is None + assert sfv == make_sfv() + assert sfv != make_sfv(**changes) + assert make_sfv(**changes) != sfv + + +def test_stream_feature_view_equality_with_udf(): + entity = Entity(name="driver_entity", join_keys=["test_key"]) + stream_source = KafkaSource( + name="kafka", + timestamp_field="event_timestamp", + kafka_bootstrap_servers="", + message_format=AvroFormat(""), + topic="topic", + batch_source=FileSource(path="some path"), + ) + + def udf_a(df): + return df + + def udf_b(df): + return df.head(1) + + def make_sfv(udf): + return StreamFeatureView( + name="test kafka stream feature view", + entities=[entity], + ttl=timedelta(days=30), + schema=[Field(name="dummy_field", dtype=Float32)], + source=stream_source, + timestamp_field="event_timestamp", + mode="pandas", + udf=udf, + ) + + assert make_sfv(udf_a) == make_sfv(udf_a) + assert make_sfv(udf_a) != make_sfv(None) + assert make_sfv(None) != make_sfv(udf_a) + assert make_sfv(udf_a) != make_sfv(udf_b)