Skip to content

Commit bbaca76

Browse files
jiwidishuchu
jiwidi
and
shuchu
authored
fix: Break event timestamp ties by created timestamp in Milvus write batches (#7084)
* fix: Break event timestamp ties by created timestamp in Milvus write batches A batch with two rows for one entity and the same event timestamp kept the first row, whatever its created timestamp. The offline stores break such ties on the created timestamp, and the Redis online store keeps the last row it writes. The Milvus store now keeps the row with the latest event timestamp, then the latest created timestamp, then the last row of the batch, so a correction with a newer created timestamp is not lost. Signed-off-by: jiwidi <[email protected]> * test: Cover a Milvus write batch with several entities Signed-off-by: jiwidi <[email protected]> --------- Signed-off-by: jiwidi <[email protected]> Co-authored-by: Shuchu Han <[email protected]>
1 parent 68de84e commit bbaca76

3 files changed

Lines changed: 102 additions & 4 deletions

File tree

‎sdk/python/feast/infra/online_stores/milvus_online_store/milvus.py‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -567,10 +567,12 @@ def online_write_batch(
567567
single_entity_record[field] = _default_for_milvus_type(
568568
field_type
569569
)
570-
# Store only the latest event timestamp per entity
571-
if (
572-
entity_key_str not in unique_entities
573-
or unique_entities[entity_key_str]["event_ts"] < timestamp_int
570+
# Keep one row per entity: the latest event timestamp wins, then the
571+
# latest created timestamp, then the last row of the batch.
572+
previous = unique_entities.get(entity_key_str)
573+
if previous is None or (previous["event_ts"], previous["created_ts"]) <= (
574+
timestamp_int,
575+
created_ts_int,
574576
):
575577
unique_entities[entity_key_str] = single_entity_record
576578

‎sdk/python/tests/integration/online_store/test_milvus_remote.py‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,39 @@ def test_scalar_feature_view_round_trip(
166166
assert rows[0] is not None and rows[0]["city"].string_val == "Paris"
167167

168168

169+
def test_write_batch_tie_break(
170+
tmp_path: Path, project: str, store: MilvusOnlineStore
171+
) -> None:
172+
"""Equal event timestamps: the row with the latest created timestamp wins."""
173+
config = _repo_config(tmp_path, project, consistency_level="Strong")
174+
fv = _scalar_feature_view()
175+
store.update(config, [], [fv], [], [], partial=False)
176+
base = datetime(2024, 1, 1, tzinfo=timezone.utc)
177+
178+
store.online_write_batch(
179+
config,
180+
fv,
181+
[
182+
(
183+
_entity_key(1),
184+
{"city": ValueProto(string_val="Rome")},
185+
base,
186+
base + timedelta(seconds=2),
187+
),
188+
(
189+
_entity_key(1),
190+
{"city": ValueProto(string_val="Oslo")},
191+
base,
192+
base + timedelta(seconds=1),
193+
),
194+
],
195+
progress=None,
196+
)
197+
198+
rows = _read(store, config, fv, [1], ["city"])
199+
assert rows[0] is not None and rows[0]["city"].string_val == "Rome"
200+
201+
169202
def _vector_feature_view(name: str = "driver_embeddings") -> FeatureView:
170203
return FeatureView(
171204
name=name,

‎sdk/python/tests/unit/infra/online_store/test_milvus_online_store.py‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,69 @@ def test_scalar_feature_view_round_trip_with_placeholder(tmp_path: Path) -> None
160160
assert placeholder["params"]["dim"] == PLACEHOLDER_VECTOR_DIM
161161

162162

163+
@pytest.mark.parametrize(
164+
"rows, expected_city",
165+
[
166+
# Equal event timestamps: the latest created timestamp wins, in any order.
167+
([("Oslo", 0, 1), ("Rome", 0, 2)], "Rome"),
168+
([("Rome", 0, 2), ("Oslo", 0, 1)], "Rome"),
169+
# A newer event timestamp wins over a newer created timestamp.
170+
([("Oslo", 1, 1), ("Rome", 0, 2)], "Oslo"),
171+
# Equal timestamps: the last row of the batch wins.
172+
([("Oslo", 0, 1), ("Rome", 0, 1)], "Rome"),
173+
],
174+
)
175+
def test_write_batch_keeps_the_newest_row_per_entity(
176+
tmp_path: Path, rows: List[Tuple[str, int, int]], expected_city: str
177+
) -> None:
178+
store = MilvusOnlineStore()
179+
config = _lite_config(tmp_path)
180+
fv = _scalar_feature_view()
181+
base = datetime(2024, 1, 1, tzinfo=timezone.utc)
182+
183+
store.online_write_batch(
184+
config,
185+
fv,
186+
[
187+
(
188+
_entity_key(1),
189+
{"city": ValueProto(string_val=city)},
190+
base + timedelta(seconds=event_ts),
191+
base + timedelta(seconds=created_ts),
192+
)
193+
for city, event_ts, created_ts in rows
194+
],
195+
progress=None,
196+
)
197+
198+
rows_read = _read(store, config, fv, [1], ["city"])
199+
assert rows_read[0] is not None
200+
assert rows_read[0]["city"].string_val == expected_city
201+
202+
203+
def test_write_batch_keeps_one_row_per_entity(tmp_path: Path) -> None:
204+
store = MilvusOnlineStore()
205+
config = _lite_config(tmp_path)
206+
fv = _scalar_feature_view()
207+
now = datetime(2024, 1, 1, tzinfo=timezone.utc)
208+
209+
store.online_write_batch(
210+
config,
211+
fv,
212+
[
213+
(_entity_key(1), {"city": ValueProto(string_val="Oslo")}, now, now),
214+
(_entity_key(2), {"city": ValueProto(string_val="Rome")}, now, now),
215+
],
216+
progress=None,
217+
)
218+
219+
rows = _read(store, config, fv, [1, 2], ["city"])
220+
assert [row["city"].string_val for row in rows if row is not None] == [
221+
"Oslo",
222+
"Rome",
223+
]
224+
225+
163226
@patch(f"{MILVUS_MODULE}.MilvusClient")
164227
def test_placeholder_vector_is_indexed_and_valid(mock_client_cls: MagicMock) -> None:
165228
mock_client = _mock_client(mock_client_cls, has_collection=False)

0 commit comments

Comments
 (0)