Skip to content

Commit d3e2041

Browse files
jiwidishuchu
jiwidi
and
shuchu
authored
feat: Add row deletion to the Milvus online store (#7082)
* feat: Add row deletion to the Milvus online store The Milvus online store had no operation between upsert and dropping a collection, so a row could not be removed when its entity left the catalogue. The Redis online store can remove rows with delete_entity_values and delete_table. MilvusOnlineStore.online_delete deletes rows by entity key or by filter. It accepts exactly one of the two, deletes nothing for an empty key list, rejects a filter without conditions, and does not create a missing collection. It works on every collection that Feast created. Signed-off-by: jiwidi <[email protected]> * test: Cover online_delete with an empty entity key list Signed-off-by: jiwidi <[email protected]> * fix: Reject an empty group anywhere in the Milvus deletion filter The Milvus filter translator drops an empty nested group, so a filter such as and(brand_id == 'acme', or()) deleted every acme row instead of none. online_delete now rejects a filter with an empty group at any level before it connects. The delete path loads the collection with _ensure_loaded instead of _get_or_create_collection, which could not create one after the has_collection guard, and reuses the collection name it computed. Signed-off-by: jiwidi <[email protected]> * ci: Re-run the checks after a macOS runner timeout Signed-off-by: jiwidi <[email protected]> --------- Signed-off-by: jiwidi <[email protected]> Co-authored-by: Shuchu Han <[email protected]>
1 parent 147adad commit d3e2041

4 files changed

Lines changed: 197 additions & 1 deletion

File tree

‎docs/reference/online-stores/milvus.md‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -197,6 +197,26 @@ When Feast finds an existing collection it checks its load state and loads it on
197197
Reads and searches never load collections, so a collection released outside Feast is only reloaded
198198
the next time a Feast process first accesses it.
199199

200+
## Row deletion
201+
202+
`MilvusOnlineStore.online_delete` removes rows of a feature view, by entity key or by filter:
203+
204+
```python
205+
from feast.filter_models import ComparisonFilter
206+
207+
online_store.online_delete(config, feature_view, entity_keys=[entity_key])
208+
online_store.online_delete(
209+
config,
210+
feature_view,
211+
filters=ComparisonFilter(type="eq", key="brand_id", value="acme"),
212+
)
213+
```
214+
215+
Give exactly one of `entity_keys` and `filters`. An empty key list deletes nothing, and a filter
216+
with an empty group is rejected, so a mistake cannot delete more rows than it names. A deleted row
217+
leaves no tombstone: a later write creates it again, and Feast does not order a delete against
218+
writes from other processes. Deletion works on every collection that Feast created.
219+
200220
## Feature views without vectors
201221

202222
Milvus requires every collection to have a vector field. For feature views that have no vector

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

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,13 @@ def translate_compound(self, f: CompoundFilter) -> str:
202202
return operator.join(clauses)
203203

204204

205+
def _has_empty_group(filters: FilterType) -> bool:
206+
"""True when a compound filter has no conditions, at any level."""
207+
return isinstance(filters, CompoundFilter) and (
208+
not filters.filters or any(_has_empty_group(f) for f in filters.filters)
209+
)
210+
211+
205212
class MilvusOnlineStoreConfig(FeastConfigBaseModel, VectorStoreConfig):
206213
"""
207214
Configuration for the Milvus online store.
@@ -706,6 +713,52 @@ def online_read(
706713

707714
return result_list
708715

716+
def online_delete(
717+
self,
718+
config: RepoConfig,
719+
table: FeatureView,
720+
entity_keys: Optional[List[EntityKeyProto]] = None,
721+
*,
722+
filters: Optional[Union[ComparisonFilter, CompoundFilter]] = None,
723+
) -> None:
724+
"""Delete rows of a feature view by entity key or by filter.
725+
726+
Give exactly one of ``entity_keys`` and ``filters``. An empty key list
727+
deletes nothing. A filter with an empty group is rejected, so a mistake
728+
cannot delete more rows than it names. No tombstone is kept: a later
729+
write creates the row again.
730+
"""
731+
if (entity_keys is None) == (filters is None):
732+
raise ValueError("Give exactly one of entity_keys or filters")
733+
selector: Dict[str, Any]
734+
if entity_keys is not None:
735+
if not entity_keys:
736+
return
737+
selector = {
738+
"ids": [
739+
serialize_entity_key(
740+
entity_key,
741+
entity_key_serialization_version=config.entity_key_serialization_version,
742+
).hex()
743+
for entity_key in entity_keys
744+
]
745+
}
746+
else:
747+
if _has_empty_group(filters):
748+
raise ValueError(
749+
"Each group of the deletion filter must have a condition"
750+
)
751+
selector = {"filter": MilvusFilterTranslator().translate(filters)}
752+
self.client = self._connect(config)
753+
collection_name = _table_id(
754+
config.project, table, config.registry.enable_online_feature_view_versioning
755+
)
756+
if collection_name not in self._collections:
757+
if not self.client.has_collection(collection_name):
758+
return
759+
self._ensure_loaded(collection_name)
760+
self.client.delete(collection_name=collection_name, **selector)
761+
709762
def update(
710763
self,
711764
config: RepoConfig,

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

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -407,3 +407,26 @@ def test_partition_key_filtering(
407407
filters=ComparisonFilter(type="eq", key="brand_id", value="globex"),
408408
)
409409
assert [hit["city"].string_val for hit in hits] == ["Rome"]
410+
411+
412+
def test_online_delete(tmp_path: Path, project: str, store: MilvusOnlineStore) -> None:
413+
config = _repo_config(tmp_path, project, consistency_level="Strong")
414+
fv = _vector_feature_view()
415+
store.update(config, [], [fv], [], [], partial=False)
416+
_write_rows(store, config, fv, _vector_rows())
417+
418+
store.online_delete(config, fv, [_entity_key(1)])
419+
assert [row is None for row in _read(store, config, fv, [1, 2], ["city"])] == [
420+
True,
421+
False,
422+
]
423+
424+
store.online_delete(
425+
config, fv, filters=ComparisonFilter(type="eq", key="city", value="Rome")
426+
)
427+
assert _read(store, config, fv, [1, 2], ["city"]) == [None, None]
428+
429+
# No tombstone is kept, so the row can be written again.
430+
_write_rows(store, config, fv, {1: _vector_rows()[1]})
431+
rows = _read(store, config, fv, [1], ["city"])
432+
assert rows[0] is not None and rows[0]["city"].string_val == "Paris"

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

Lines changed: 101 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212

1313
from feast import Entity, FeatureView
1414
from feast.field import Field
15-
from feast.filter_models import ComparisonFilter
15+
from feast.filter_models import ComparisonFilter, CompoundFilter
1616
from feast.infra.online_stores.milvus_online_store.milvus import (
1717
PLACEHOLDER_VECTOR_DIM,
1818
PLACEHOLDER_VECTOR_FIELD,
@@ -746,3 +746,103 @@ def test_warns_when_existing_collection_lacks_partition_key(
746746

747747
assert "created without partition key 'brand_id'" in caplog.text
748748
mock_client.create_collection.assert_not_called()
749+
750+
751+
def test_online_delete_by_entity_key_and_by_filter(tmp_path: Path) -> None:
752+
config = _lite_config(tmp_path)
753+
fv = _scalar_feature_view()
754+
store = MilvusOnlineStore()
755+
store.update(config, [], [fv], [], [], partial=False)
756+
_write_rows(
757+
store,
758+
config,
759+
fv,
760+
{
761+
driver_id: {
762+
"trips_today": ValueProto(float_val=float(driver_id)),
763+
"city": ValueProto(string_val=city),
764+
}
765+
for driver_id, city in [(1, "Paris"), (2, "Rome"), (3, "Rome")]
766+
},
767+
)
768+
769+
store.online_delete(config, fv, [_entity_key(1)])
770+
rows = _read(store, config, fv, [1, 2, 3], ["city"])
771+
assert [row is None for row in rows] == [True, False, False]
772+
773+
store.online_delete(
774+
config, fv, filters=ComparisonFilter(type="eq", key="city", value="Rome")
775+
)
776+
assert _read(store, config, fv, [1, 2, 3], ["city"]) == [None, None, None]
777+
778+
# No tombstone is kept, so the row can be written again.
779+
_write_rows(
780+
store,
781+
config,
782+
fv,
783+
{
784+
1: {
785+
"trips_today": ValueProto(float_val=4.0),
786+
"city": ValueProto(string_val="Oslo"),
787+
}
788+
},
789+
)
790+
rows = _read(store, config, fv, [1], ["city"])
791+
assert rows[0] is not None and rows[0]["city"].string_val == "Oslo"
792+
793+
794+
@pytest.mark.parametrize(
795+
"kwargs",
796+
[
797+
{},
798+
{
799+
"entity_keys": [_entity_key(1)],
800+
"filters": ComparisonFilter(type="eq", key="city", value="Rome"),
801+
},
802+
{"filters": CompoundFilter(type="and", filters=[])},
803+
{
804+
"filters": CompoundFilter(
805+
type="and",
806+
filters=[
807+
ComparisonFilter(type="eq", key="brand_id", value="acme"),
808+
CompoundFilter(type="or", filters=[]),
809+
],
810+
)
811+
},
812+
],
813+
)
814+
def test_online_delete_rejects_ambiguous_and_unbounded_calls(
815+
kwargs: Dict[str, Any],
816+
) -> None:
817+
store = MilvusOnlineStore()
818+
store.client = MagicMock()
819+
820+
with pytest.raises(ValueError):
821+
store.online_delete(_mock_config(), _scalar_feature_view(), **kwargs)
822+
823+
store.client.delete.assert_not_called()
824+
825+
826+
@patch(f"{MILVUS_MODULE}.MilvusClient")
827+
def test_online_delete_does_not_create_a_missing_collection(
828+
mock_client_cls: MagicMock,
829+
) -> None:
830+
mock_client = _mock_client(mock_client_cls, has_collection=False)
831+
store = MilvusOnlineStore()
832+
833+
store.online_delete(_mock_config(), _scalar_feature_view(), [_entity_key(1)])
834+
store.online_delete(_mock_config(), _scalar_feature_view(), [])
835+
836+
mock_client.create_collection.assert_not_called()
837+
mock_client.delete.assert_not_called()
838+
839+
840+
@patch(f"{MILVUS_MODULE}.MilvusClient")
841+
def test_online_delete_with_no_entity_keys_sends_nothing(
842+
mock_client_cls: MagicMock,
843+
) -> None:
844+
MilvusOnlineStore().online_delete(
845+
_mock_config(), _scalar_feature_view(), entity_keys=[]
846+
)
847+
848+
mock_client_cls.assert_not_called()

0 commit comments

Comments
 (0)