Skip to content

Commit 2d401f9

Browse files
jang-hsclaude
jang-hs
and
claude
committed
fix: route only each backend's own FeatureViews in HybridOnlineStore
Backends act on every FeatureView handed to them, so passing the full keep/delete lists made e.g. SQLite create tables for Redis-routed views, and teardown let each backend drop another backend's tables. - group FeatureViews by routing tag once, call each backend with only its own subset - update() now also reaches a backend that has deletions but no kept views; previously it was never called and its tables were never dropped - teardown() drops the (tribe, type, id(conf)) dedup set: grouping already gives one call per backend, and the old dedup skipped the second view routed to the same backend Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Jade <retrorca@gmail.com>
1 parent dd428a5 commit 2d401f9

2 files changed

Lines changed: 148 additions & 45 deletions

File tree

‎sdk/python/feast/infra/online_stores/hybrid_online_store/hybrid_online_store.py‎

Lines changed: 67 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -167,6 +167,41 @@ def _get_routing_tag_value(self, table: FeatureView, config: RepoConfig):
167167
tag_name = getattr(config.online_store, "routing_tag", "tribe")
168168
return table.tags.get(tag_name)
169169

170+
def _group_by_routing_tag(
171+
self,
172+
tables: Sequence[FeatureView],
173+
config: RepoConfig,
174+
require_tag: bool,
175+
) -> Dict[str, List[FeatureView]]:
176+
"""
177+
Bucket FeatureViews by the (lower-cased) value of their routing tag.
178+
179+
Backends act on every FeatureView handed to them, so each one must only
180+
ever see its own bucket. Passing the full list made e.g. SQLite create
181+
tables for Redis-routed views.
182+
183+
Args:
184+
tables: FeatureViews to group.
185+
config: Feast RepoConfig.
186+
require_tag: Raise on an untagged FeatureView instead of skipping it.
187+
Returns:
188+
Mapping of tag value to the FeatureViews carrying it.
189+
Raises:
190+
ValueError: If ``require_tag`` and a FeatureView has no routing tag.
191+
"""
192+
grouped: Dict[str, List[FeatureView]] = {}
193+
for table in tables:
194+
tribe = self._get_routing_tag_value(table, config)
195+
if not tribe:
196+
if require_tag:
197+
tag_name = getattr(config.online_store, "routing_tag", "tribe")
198+
raise ValueError(
199+
f"FeatureView must have a '{tag_name}' tag to use HybridOnlineStore."
200+
)
201+
continue
202+
grouped.setdefault(tribe.lower(), []).append(table)
203+
return grouped
204+
170205
def online_write_batch(
171206
self,
172207
config: RepoConfig,
@@ -273,31 +308,32 @@ def update(
273308
ValueError: If a FeatureView does not have the required tag.
274309
NotImplementedError: If no online store is found for a tag value.
275310
"""
276-
for table in tables_to_keep:
277-
tribe = self._get_routing_tag_value(table, config)
278-
if not tribe:
279-
tag_name = getattr(config.online_store, "routing_tag", "tribe")
280-
raise ValueError(
281-
f"FeatureView must have a '{tag_name}' tag to use HybridOnlineStore."
282-
)
311+
keep_by_tribe = self._group_by_routing_tag(
312+
tables_to_keep, config, require_tag=True
313+
)
314+
# Untagged views on the way out are skipped rather than fatal: they may
315+
# predate the routing tag, and there is no backend to route them to.
316+
delete_by_tribe = self._group_by_routing_tag(
317+
tables_to_delete, config, require_tag=False
318+
)
319+
for tribe in {**keep_by_tribe, **delete_by_tribe}:
283320
online_store = self._get_online_store(tribe, config)
284-
if online_store:
285-
# Local name: rebinding `config` here would feed the next
286-
# iteration the selected backend's config instead of the hybrid
287-
# one, losing `routing_tag` from the second FeatureView on.
288-
store_config = RepoConfig(**self._prepare_repo_conf(config, tribe))
289-
online_store.update(
290-
store_config,
291-
tables_to_delete,
292-
tables_to_keep,
293-
entities_to_delete,
294-
entities_to_keep,
295-
partial,
296-
)
297-
else:
321+
if not online_store:
298322
raise NotImplementedError(
299323
f"No online store found for {getattr(config.online_store, 'routing_tag', 'tribe')} tag '{tribe}'. Please check your configuration."
300324
)
325+
# Local name: rebinding `config` here would feed the next iteration
326+
# the selected backend's config instead of the hybrid one, losing
327+
# `routing_tag`.
328+
store_config = RepoConfig(**self._prepare_repo_conf(config, tribe))
329+
online_store.update(
330+
store_config,
331+
delete_by_tribe.get(tribe, []),
332+
keep_by_tribe.get(tribe, []),
333+
entities_to_delete,
334+
entities_to_keep,
335+
partial,
336+
)
301337

302338
def teardown(
303339
self,
@@ -313,27 +349,13 @@ def teardown(
313349
tables: Sequence of FeatureViews to teardown.
314350
entities: Sequence of Entities to teardown.
315351
"""
316-
# Use a set of (tribe, store_type, conf_id) to avoid duplicate teardowns for the same instance
317-
tribes_seen = set()
318-
online_stores_cfg = getattr(config.online_store, "online_stores", [])
319-
tag_name = getattr(config.online_store, "routing_tag", "tribe")
320-
for table in tables:
321-
tribe = table.tags.get(tag_name)
322-
if not tribe:
323-
continue
324-
# Find all store configs matching this tribe (supporting multiple instances of the same type)
325-
for store_cfg in online_stores_cfg:
326-
store_type = store_cfg.type
327-
# Use id(store_cfg.conf) to distinguish different configs of the same type
328-
key = (tribe, store_type, id(store_cfg.conf))
329-
if key in tribes_seen:
330-
continue
331-
tribes_seen.add(key)
332-
# Only select the online store if tribe matches the type (or you can add a mapping in config for more flexibility)
333-
if tribe.lower() == store_type.split(".")[-1].lower():
334-
online_store = self._get_online_store(tribe, config)
335-
if online_store:
336-
store_config = RepoConfig(
337-
**self._prepare_repo_conf(config, tribe)
338-
)
339-
online_store.teardown(store_config, tables, entities)
352+
# Grouping both dedupes backends and keeps each one from tearing down
353+
# another backend's FeatureViews. Untagged views have no backend to
354+
# route to and are skipped, as before.
355+
for tribe, tribe_tables in self._group_by_routing_tag(
356+
tables, config, require_tag=False
357+
).items():
358+
online_store = self._get_online_store(tribe, config)
359+
if online_store:
360+
store_config = RepoConfig(**self._prepare_repo_conf(config, tribe))
361+
online_store.teardown(store_config, tribe_tables, entities)

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

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,3 +100,84 @@ def test_update_routes_every_feature_view(repo_config, entity):
100100
assert redis_update.call_count == 1
101101
assert sqlite_update.call_count == 1
102102
assert repo_config.online_store.routing_tag == ROUTING_TAG
103+
104+
105+
def test_update_passes_each_backend_only_its_own_tables(repo_config, entity):
106+
"""A backend must not create or drop infrastructure for another backend's views."""
107+
fv_redis = _feature_view("fv_redis", "redis", entity)
108+
fv_sqlite = _feature_view("fv_sqlite", "sqlite", entity)
109+
fv_redis_gone = _feature_view("fv_redis_gone", "redis", entity)
110+
111+
with (
112+
patch(
113+
"feast.infra.online_stores.redis.RedisOnlineStore.update"
114+
) as redis_update,
115+
patch(
116+
"feast.infra.online_stores.sqlite.SqliteOnlineStore.update"
117+
) as sqlite_update,
118+
):
119+
HybridOnlineStore().update(
120+
config=repo_config,
121+
tables_to_delete=[fv_redis_gone],
122+
tables_to_keep=[fv_redis, fv_sqlite],
123+
entities_to_delete=[],
124+
entities_to_keep=[entity],
125+
partial=False,
126+
)
127+
128+
_, redis_delete, redis_keep, *_ = redis_update.call_args.args
129+
_, sqlite_delete, sqlite_keep, *_ = sqlite_update.call_args.args
130+
assert redis_keep == [fv_redis]
131+
assert redis_delete == [fv_redis_gone]
132+
assert sqlite_keep == [fv_sqlite]
133+
assert sqlite_delete == []
134+
135+
136+
def test_update_reaches_a_backend_with_only_deletions(repo_config, entity):
137+
"""A backend whose views are all being removed still needs its update() call."""
138+
with (
139+
patch(
140+
"feast.infra.online_stores.redis.RedisOnlineStore.update"
141+
) as redis_update,
142+
patch(
143+
"feast.infra.online_stores.sqlite.SqliteOnlineStore.update"
144+
) as sqlite_update,
145+
):
146+
HybridOnlineStore().update(
147+
config=repo_config,
148+
tables_to_delete=[_feature_view("fv_redis_gone", "redis", entity)],
149+
tables_to_keep=[_feature_view("fv_sqlite", "sqlite", entity)],
150+
entities_to_delete=[],
151+
entities_to_keep=[entity],
152+
partial=False,
153+
)
154+
155+
assert redis_update.call_count == 1
156+
assert sqlite_update.call_count == 1
157+
158+
159+
def test_teardown_passes_each_backend_only_its_own_tables(repo_config, entity):
160+
"""Teardown used to hand every backend the full table list."""
161+
fv_redis = _feature_view("fv_redis", "redis", entity)
162+
fv_sqlite = _feature_view("fv_sqlite", "sqlite", entity)
163+
fv_sqlite2 = _feature_view("fv_sqlite2", "sqlite", entity)
164+
165+
with (
166+
patch(
167+
"feast.infra.online_stores.redis.RedisOnlineStore.teardown"
168+
) as redis_teardown,
169+
patch(
170+
"feast.infra.online_stores.sqlite.SqliteOnlineStore.teardown"
171+
) as sqlite_teardown,
172+
):
173+
HybridOnlineStore().teardown(
174+
config=repo_config,
175+
tables=[fv_redis, fv_sqlite, fv_sqlite2],
176+
entities=[entity],
177+
)
178+
179+
assert redis_teardown.call_count == 1
180+
assert sqlite_teardown.call_count == 1
181+
assert redis_teardown.call_args.args[1] == [fv_redis]
182+
# Both sqlite views in one call: the old dedup dropped the second one.
183+
assert sqlite_teardown.call_args.args[1] == [fv_sqlite, fv_sqlite2]

0 commit comments

Comments
 (0)