Visitar URL original
fix: Preserve an unset aggregation time window across a proto round-trip · feast-dev/feast@1bd6944 · GitHub
Skip to content

Commit 1bd6944

Browse files
Rodrigo-Palmantkathole
Rodrigo-Palma
authored and
ntkathole
committed
fix: Preserve an unset aggregation time window across a proto round-trip
Aggregation.__init__ types time_window and slide_interval as Optional and to_proto honours that: it only writes the Duration when the value is not None. from_proto tests the value instead of the presence, and an absent Duration also reports ToNanoseconds() == 0, so an aggregation defined without a window comes back with timedelta(0). That difference is load bearing. aggregation_specs_to_agg_ops rejects a windowed aggregation in online serving with 'if getattr(agg, "time_window", None) is not None', so an on demand feature view with a plain Aggregation(column=..., function=...) works in process and then fails with 'Time window aggregation is not supported in online serving.' on the first get_online_features after feast apply, whatever the registry. The same round-trip runs for StreamFeatureView. Read the field through HasField, which is available because both are google.protobuf.Duration message fields. Aggregation.__eq__ compares these two attributes, so this also stops a registry diff from seeing an unchanged feature view as modified. Signed-off-by: Rodrigo-Palma <email.rodrigopalma@gmail.com>
1 parent 139704f commit 1bd6944

2 files changed

Lines changed: 48 additions & 6 deletions

File tree

‎sdk/python/feast/aggregation/__init__.py‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -69,15 +69,15 @@ def to_proto(self) -> AggregationProto:
6969
@classmethod
7070
def from_proto(cls, agg_proto: AggregationProto):
7171
time_window = (
72-
timedelta(days=0)
73-
if agg_proto.time_window.ToNanoseconds() == 0
74-
else agg_proto.time_window.ToTimedelta()
72+
agg_proto.time_window.ToTimedelta()
73+
if agg_proto.HasField("time_window")
74+
else None
7575
)
7676

7777
slide_interval = (
78-
timedelta(days=0)
79-
if agg_proto.slide_interval.ToNanoseconds() == 0
80-
else agg_proto.slide_interval.ToTimedelta()
78+
agg_proto.slide_interval.ToTimedelta()
79+
if agg_proto.HasField("slide_interval")
80+
else None
8181
)
8282
aggregation = cls(
8383
column=agg_proto.column,

‎sdk/python/tests/unit/test_on_demand_feature_view_aggregation.py‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,9 @@
1717
import pyarrow as pa
1818

1919
from feast.aggregation import Aggregation
20+
from feast.field import Field
21+
from feast.on_demand_feature_view import OnDemandFeatureView, on_demand_feature_view
22+
from feast.types import Float32, Int64
2023
from feast.utils import _apply_aggregations_to_response
2124

2225

@@ -87,3 +90,42 @@ def test_empty_data_returns_empty():
8790
result = _apply_aggregations_to_response(data, aggs, ["driver_id"], "python")
8891

8992
assert result == data
93+
94+
95+
def test_aggregation_without_time_window_survives_proto_roundtrip():
96+
"""An Aggregation with no time window keeps None through to_proto/from_proto."""
97+
agg = Aggregation(column="trips", function="sum")
98+
assert agg.time_window is None
99+
assert agg.slide_interval is None
100+
101+
restored = Aggregation.from_proto(agg.to_proto())
102+
103+
assert restored.time_window is None
104+
assert restored.slide_interval is None
105+
assert restored == agg
106+
107+
108+
def test_odfv_aggregation_still_serves_online_after_proto_roundtrip():
109+
"""A registry round-trip must not turn an unset window into a zero window."""
110+
111+
@on_demand_feature_view(
112+
sources=[],
113+
input_schema=[
114+
Field(name="driver_id", dtype=Int64),
115+
Field(name="trips", dtype=Int64),
116+
],
117+
schema=[Field(name="sum_trips", dtype=Float32)],
118+
aggregations=[Aggregation(column="trips", function="sum")],
119+
mode="python",
120+
)
121+
def agg_view(inputs):
122+
return {"sum_trips": inputs["sum_trips"]}
123+
124+
restored = OnDemandFeatureView.from_proto(agg_view.to_proto())
125+
126+
data = {"driver_id": [1, 1, 2, 2], "trips": [10, 20, 15, 25]}
127+
result = _apply_aggregations_to_response(
128+
data, restored.aggregations, ["driver_id"], "python"
129+
)
130+
131+
assert result == {"driver_id": [1, 2], "sum_trips": [30, 40]}

0 commit comments

Comments
 (0)