From bfcfeee7776e5f8879bdb148c8cfff3b110dcb4c Mon Sep 17 00:00:00 2001 From: Vidit Patankar Date: Sun, 18 Oct 2026 14:08:18 +0530 Subject: [PATCH 1/2] fix: Compare aggregations, mode and tiling in StreamFeatureView equality StreamFeatureView.__eq__ returned early when neither view had a udf, so mode, timestamp_field, udf_string and aggregations were never compared for UDF-less stream feature views, and enable_tiling / tiling_hop_size were not compared at all. The SQL registry relies on this equality to skip unchanged writes and to detect version pin conflicts. Compare these fields regardless of the udf, normalizing mode and udf_string the way to_proto() writes them so that a view still equals its from_proto(to_proto()) round trip. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Vidit Patankar --- sdk/python/feast/stream_feature_view.py | 13 +++-- .../tests/unit/test_stream_feature_view.py | 55 +++++++++++++++++++ 2 files changed, 62 insertions(+), 6 deletions(-) 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..8c82b2b688d 100644 --- a/sdk/python/tests/unit/test_stream_feature_view.py +++ b/sdk/python/tests/unit/test_stream_feature_view.py @@ -345,3 +345,58 @@ 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 From 970e2fdea64d3c030e7b7a6320830a2ee2538a6e Mon Sep 17 00:00:00 2001 From: Vidit Patankar Date: Sun, 18 Oct 2026 15:00:23 +0530 Subject: [PATCH 2/2] test: Cover UDF branches of StreamFeatureView equality Exercise the cases where only one view has a udf and where both have different udfs, which were not covered by the UDF-less equality test. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Vidit Patankar --- .../tests/unit/test_stream_feature_view.py | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/sdk/python/tests/unit/test_stream_feature_view.py b/sdk/python/tests/unit/test_stream_feature_view.py index 8c82b2b688d..c585719096b 100644 --- a/sdk/python/tests/unit/test_stream_feature_view.py +++ b/sdk/python/tests/unit/test_stream_feature_view.py @@ -400,3 +400,38 @@ def make_sfv(**kwargs): 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)