From dd428a5349e9ef7872384de32418da4eb897e6fd Mon Sep 17 00:00:00 2001 From: Jade Date: Fri, 21 Aug 2026 10:59:52 +0900 Subject: [PATCH 1/2] fix: keep HybridOnlineStore routing_tag across FeatureViews MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HybridOnlineStore.update() applied the first FeatureView and then failed on the second with ValueError: FeatureView must have a 'tribe' tag to use HybridOnlineStore no matter how the views were tagged. The message reports 'tribe' — the fallback, not the configured routing_tag — because config.online_store is no longer the HybridOnlineStoreConfig by the time the second view is read. Two causes: 1. _prepare_repo_conf mutated the caller's RepoConfig. `rconfig = config` is an alias, so both the attribute assignment and the __dict__ writes landed on the caller's object and replaced online_store with the selected backend's config. It also injected `type` into online_store.conf, which belongs to the user's config. 2. update() and teardown() rebound the `config` parameter inside their loops, so the next iteration read routing_tag off the narrowed config even once the mutation was gone. Build the returned mapping from a copy, construct a new dict for the backend conf, and assign the per-backend RepoConfig to a local name in the loops. Behaviour is otherwise unchanged: the returned kwargs are identical to what the mutating version produced. Adds a unit regression test covering both the multi-FeatureView routing and the caller's config staying intact. Both fail on master. Signed-off-by: Jade --- .../hybrid_online_store.py | 30 ++++-- .../online_store/test_hybrid_online_store.py | 102 ++++++++++++++++++ 2 files changed, 123 insertions(+), 9 deletions(-) create mode 100644 sdk/python/tests/unit/infra/online_store/test_hybrid_online_store.py diff --git a/sdk/python/feast/infra/online_stores/hybrid_online_store/hybrid_online_store.py b/sdk/python/feast/infra/online_stores/hybrid_online_store/hybrid_online_store.py index 8faefdbd344..02a54d0a30b 100644 --- a/sdk/python/feast/infra/online_stores/hybrid_online_store/hybrid_online_store.py +++ b/sdk/python/feast/infra/online_stores/hybrid_online_store/hybrid_online_store.py @@ -139,21 +139,28 @@ def _prepare_repo_conf(self, config: RepoConfig, online_store_type: str): """ Prepare a RepoConfig for the selected online store backend. + The caller's RepoConfig is left untouched: the returned mapping is built + from a copy. Mutating it in place replaced the caller's HybridOnlineStore + config with the selected backend's config, so every later lookup of + ``routing_tag`` fell back to the "tribe" default. + Args: config: The original Feast RepoConfig. online_store_type: The type of the online store backend to use. Returns: A dictionary representing the updated RepoConfig for the selected backend. """ - rconfig = config + online_config = config.online_config for online_store in config.online_store.online_stores: if online_store.type.split(".")[-1].lower() == online_store_type.lower(): - rconfig.online_config = online_store.conf - rconfig.online_config["type"] = online_store.type - data = rconfig.__dict__ + # New dict rather than a mutated one: ``conf`` belongs to the + # caller's config and is read again on the next lookup. + online_config = {**online_store.conf, "type": online_store.type} + data = dict(config.__dict__) data["registry"] = data["registry_config"] data["offline_store"] = data["offline_config"] - data["online_store"] = data["online_config"] + data["online_config"] = online_config + data["online_store"] = online_config return data def _get_routing_tag_value(self, table: FeatureView, config: RepoConfig): @@ -275,9 +282,12 @@ def update( ) online_store = self._get_online_store(tribe, config) if online_store: - config = RepoConfig(**self._prepare_repo_conf(config, tribe)) + # Local name: rebinding `config` here would feed the next + # iteration the selected backend's config instead of the hybrid + # one, losing `routing_tag` from the second FeatureView on. + store_config = RepoConfig(**self._prepare_repo_conf(config, tribe)) online_store.update( - config, + store_config, tables_to_delete, tables_to_keep, entities_to_delete, @@ -323,5 +333,7 @@ def teardown( if tribe.lower() == store_type.split(".")[-1].lower(): online_store = self._get_online_store(tribe, config) if online_store: - config = RepoConfig(**self._prepare_repo_conf(config, tribe)) - online_store.teardown(config, tables, entities) + store_config = RepoConfig( + **self._prepare_repo_conf(config, tribe) + ) + online_store.teardown(store_config, tables, entities) diff --git a/sdk/python/tests/unit/infra/online_store/test_hybrid_online_store.py b/sdk/python/tests/unit/infra/online_store/test_hybrid_online_store.py new file mode 100644 index 00000000000..f633652009a --- /dev/null +++ b/sdk/python/tests/unit/infra/online_store/test_hybrid_online_store.py @@ -0,0 +1,102 @@ +from unittest.mock import patch + +import pytest + +from feast import Entity, FeatureView, Field, FileSource, RepoConfig, ValueType +from feast.infra.online_stores.hybrid_online_store.hybrid_online_store import ( + HybridOnlineStore, + HybridOnlineStoreConfig, +) +from feast.types import PrimitiveFeastType + +ROUTING_TAG = "backend" + + +@pytest.fixture +def entity(): + return Entity(name="id", join_keys=["id"], value_type=ValueType.INT64) + + +def _feature_view(name: str, backend: str, entity: Entity) -> FeatureView: + return FeatureView( + name=name, + entities=[entity], + schema=[Field(name="feature1", dtype=PrimitiveFeastType.INT64)], + online=True, + tags={ROUTING_TAG: backend}, + source=FileSource( + path="/tmp/feast_hybrid_test.parquet", + event_timestamp_column="event_timestamp", + ), + ) + + +@pytest.fixture +def repo_config(): + return RepoConfig( + registry="test-registry.db", + project="test_project", + provider="local", + online_store=HybridOnlineStoreConfig( + routing_tag=ROUTING_TAG, + online_stores=[ + HybridOnlineStoreConfig.OnlineStoresWithConfig( + type="redis", + conf={"redis_type": "redis", "connection_string": "localhost:6379"}, + ), + HybridOnlineStoreConfig.OnlineStoresWithConfig( + type="sqlite", + conf={"path": "/tmp/feast_hybrid_test.db"}, + ), + ], + ), + offline_store=None, + ) + + +def test_prepare_repo_conf_does_not_mutate_caller_config(repo_config): + """The selected backend's config must not leak back into the caller's config.""" + original_online_store = repo_config.online_store + original_redis_conf = dict(repo_config.online_store.online_stores[0].conf) + + HybridOnlineStore()._prepare_repo_conf(repo_config, "redis") + + assert repo_config.online_store is original_online_store + assert repo_config.online_store.routing_tag == ROUTING_TAG + # `type` used to be injected into the caller's own conf dict. + assert repo_config.online_store.online_stores[0].conf == original_redis_conf + + +def test_update_routes_every_feature_view(repo_config, entity): + """Regression: routing used to break from the second FeatureView onwards. + + `update()` rebound `config` to the selected backend's RepoConfig, so the next + iteration read `routing_tag` off a config that no longer had one. It fell back + to the "tribe" default, found no such tag, and raised + "FeatureView must have a 'tribe' tag to use HybridOnlineStore". + """ + tables = [ + _feature_view("fv_redis", "redis", entity), + _feature_view("fv_sqlite", "sqlite", entity), + ] + + with ( + patch( + "feast.infra.online_stores.redis.RedisOnlineStore.update" + ) as redis_update, + patch( + "feast.infra.online_stores.sqlite.SqliteOnlineStore.update" + ) as sqlite_update, + ): + HybridOnlineStore().update( + config=repo_config, + tables_to_delete=[], + tables_to_keep=tables, + entities_to_delete=[], + entities_to_keep=[entity], + partial=False, + ) + + assert redis_update.call_count == 1 + assert sqlite_update.call_count == 1 + assert repo_config.online_store.routing_tag == ROUTING_TAG From 9d8a2c119026d9e189f5fff66b9c6819a51e0e44 Mon Sep 17 00:00:00 2001 From: Jade Date: Tue, 25 Aug 2026 20:51:03 +0900 Subject: [PATCH 2/2] 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 Signed-off-by: Jade --- .../hybrid_online_store.py | 112 +++++++++++------- .../online_store/test_hybrid_online_store.py | 81 +++++++++++++ 2 files changed, 148 insertions(+), 45 deletions(-) diff --git a/sdk/python/feast/infra/online_stores/hybrid_online_store/hybrid_online_store.py b/sdk/python/feast/infra/online_stores/hybrid_online_store/hybrid_online_store.py index 02a54d0a30b..74350949142 100644 --- a/sdk/python/feast/infra/online_stores/hybrid_online_store/hybrid_online_store.py +++ b/sdk/python/feast/infra/online_stores/hybrid_online_store/hybrid_online_store.py @@ -167,6 +167,41 @@ def _get_routing_tag_value(self, table: FeatureView, config: RepoConfig): tag_name = getattr(config.online_store, "routing_tag", "tribe") return table.tags.get(tag_name) + def _group_by_routing_tag( + self, + tables: Sequence[FeatureView], + config: RepoConfig, + require_tag: bool, + ) -> Dict[str, List[FeatureView]]: + """ + Bucket FeatureViews by the (lower-cased) value of their routing tag. + + Backends act on every FeatureView handed to them, so each one must only + ever see its own bucket. Passing the full list made e.g. SQLite create + tables for Redis-routed views. + + Args: + tables: FeatureViews to group. + config: Feast RepoConfig. + require_tag: Raise on an untagged FeatureView instead of skipping it. + Returns: + Mapping of tag value to the FeatureViews carrying it. + Raises: + ValueError: If ``require_tag`` and a FeatureView has no routing tag. + """ + grouped: Dict[str, List[FeatureView]] = {} + for table in tables: + tribe = self._get_routing_tag_value(table, config) + if not tribe: + if require_tag: + tag_name = getattr(config.online_store, "routing_tag", "tribe") + raise ValueError( + f"FeatureView must have a '{tag_name}' tag to use HybridOnlineStore." + ) + continue + grouped.setdefault(tribe.lower(), []).append(table) + return grouped + def online_write_batch( self, config: RepoConfig, @@ -273,31 +308,32 @@ def update( ValueError: If a FeatureView does not have the required tag. NotImplementedError: If no online store is found for a tag value. """ - for table in tables_to_keep: - tribe = self._get_routing_tag_value(table, config) - if not tribe: - tag_name = getattr(config.online_store, "routing_tag", "tribe") - raise ValueError( - f"FeatureView must have a '{tag_name}' tag to use HybridOnlineStore." - ) + keep_by_tribe = self._group_by_routing_tag( + tables_to_keep, config, require_tag=True + ) + # Untagged views on the way out are skipped rather than fatal: they may + # predate the routing tag, and there is no backend to route them to. + delete_by_tribe = self._group_by_routing_tag( + tables_to_delete, config, require_tag=False + ) + for tribe in {**keep_by_tribe, **delete_by_tribe}: online_store = self._get_online_store(tribe, config) - if online_store: - # Local name: rebinding `config` here would feed the next - # iteration the selected backend's config instead of the hybrid - # one, losing `routing_tag` from the second FeatureView on. - store_config = RepoConfig(**self._prepare_repo_conf(config, tribe)) - online_store.update( - store_config, - tables_to_delete, - tables_to_keep, - entities_to_delete, - entities_to_keep, - partial, - ) - else: + if not online_store: raise NotImplementedError( f"No online store found for {getattr(config.online_store, 'routing_tag', 'tribe')} tag '{tribe}'. Please check your configuration." ) + # Local name: rebinding `config` here would feed the next iteration + # the selected backend's config instead of the hybrid one, losing + # `routing_tag`. + store_config = RepoConfig(**self._prepare_repo_conf(config, tribe)) + online_store.update( + store_config, + delete_by_tribe.get(tribe, []), + keep_by_tribe.get(tribe, []), + entities_to_delete, + entities_to_keep, + partial, + ) def teardown( self, @@ -313,27 +349,13 @@ def teardown( tables: Sequence of FeatureViews to teardown. entities: Sequence of Entities to teardown. """ - # Use a set of (tribe, store_type, conf_id) to avoid duplicate teardowns for the same instance - tribes_seen = set() - online_stores_cfg = getattr(config.online_store, "online_stores", []) - tag_name = getattr(config.online_store, "routing_tag", "tribe") - for table in tables: - tribe = table.tags.get(tag_name) - if not tribe: - continue - # Find all store configs matching this tribe (supporting multiple instances of the same type) - for store_cfg in online_stores_cfg: - store_type = store_cfg.type - # Use id(store_cfg.conf) to distinguish different configs of the same type - key = (tribe, store_type, id(store_cfg.conf)) - if key in tribes_seen: - continue - tribes_seen.add(key) - # Only select the online store if tribe matches the type (or you can add a mapping in config for more flexibility) - if tribe.lower() == store_type.split(".")[-1].lower(): - online_store = self._get_online_store(tribe, config) - if online_store: - store_config = RepoConfig( - **self._prepare_repo_conf(config, tribe) - ) - online_store.teardown(store_config, tables, entities) + # Grouping both dedupes backends and keeps each one from tearing down + # another backend's FeatureViews. Untagged views have no backend to + # route to and are skipped, as before. + for tribe, tribe_tables in self._group_by_routing_tag( + tables, config, require_tag=False + ).items(): + online_store = self._get_online_store(tribe, config) + if online_store: + store_config = RepoConfig(**self._prepare_repo_conf(config, tribe)) + online_store.teardown(store_config, tribe_tables, entities) diff --git a/sdk/python/tests/unit/infra/online_store/test_hybrid_online_store.py b/sdk/python/tests/unit/infra/online_store/test_hybrid_online_store.py index f633652009a..fdd5bbbd038 100644 --- a/sdk/python/tests/unit/infra/online_store/test_hybrid_online_store.py +++ b/sdk/python/tests/unit/infra/online_store/test_hybrid_online_store.py @@ -100,3 +100,84 @@ def test_update_routes_every_feature_view(repo_config, entity): assert redis_update.call_count == 1 assert sqlite_update.call_count == 1 assert repo_config.online_store.routing_tag == ROUTING_TAG + + +def test_update_passes_each_backend_only_its_own_tables(repo_config, entity): + """A backend must not create or drop infrastructure for another backend's views.""" + fv_redis = _feature_view("fv_redis", "redis", entity) + fv_sqlite = _feature_view("fv_sqlite", "sqlite", entity) + fv_redis_gone = _feature_view("fv_redis_gone", "redis", entity) + + with ( + patch( + "feast.infra.online_stores.redis.RedisOnlineStore.update" + ) as redis_update, + patch( + "feast.infra.online_stores.sqlite.SqliteOnlineStore.update" + ) as sqlite_update, + ): + HybridOnlineStore().update( + config=repo_config, + tables_to_delete=[fv_redis_gone], + tables_to_keep=[fv_redis, fv_sqlite], + entities_to_delete=[], + entities_to_keep=[entity], + partial=False, + ) + + _, redis_delete, redis_keep, *_ = redis_update.call_args.args + _, sqlite_delete, sqlite_keep, *_ = sqlite_update.call_args.args + assert redis_keep == [fv_redis] + assert redis_delete == [fv_redis_gone] + assert sqlite_keep == [fv_sqlite] + assert sqlite_delete == [] + + +def test_update_reaches_a_backend_with_only_deletions(repo_config, entity): + """A backend whose views are all being removed still needs its update() call.""" + with ( + patch( + "feast.infra.online_stores.redis.RedisOnlineStore.update" + ) as redis_update, + patch( + "feast.infra.online_stores.sqlite.SqliteOnlineStore.update" + ) as sqlite_update, + ): + HybridOnlineStore().update( + config=repo_config, + tables_to_delete=[_feature_view("fv_redis_gone", "redis", entity)], + tables_to_keep=[_feature_view("fv_sqlite", "sqlite", entity)], + entities_to_delete=[], + entities_to_keep=[entity], + partial=False, + ) + + assert redis_update.call_count == 1 + assert sqlite_update.call_count == 1 + + +def test_teardown_passes_each_backend_only_its_own_tables(repo_config, entity): + """Teardown used to hand every backend the full table list.""" + fv_redis = _feature_view("fv_redis", "redis", entity) + fv_sqlite = _feature_view("fv_sqlite", "sqlite", entity) + fv_sqlite2 = _feature_view("fv_sqlite2", "sqlite", entity) + + with ( + patch( + "feast.infra.online_stores.redis.RedisOnlineStore.teardown" + ) as redis_teardown, + patch( + "feast.infra.online_stores.sqlite.SqliteOnlineStore.teardown" + ) as sqlite_teardown, + ): + HybridOnlineStore().teardown( + config=repo_config, + tables=[fv_redis, fv_sqlite, fv_sqlite2], + entities=[entity], + ) + + assert redis_teardown.call_count == 1 + assert sqlite_teardown.call_count == 1 + assert redis_teardown.call_args.args[1] == [fv_redis] + # Both sqlite views in one call: the old dedup dropped the second one. + assert sqlite_teardown.call_args.args[1] == [fv_sqlite, fv_sqlite2]