From 8fbd3d76f9fc37db55160bf27397355d0ab1dadd Mon Sep 17 00:00:00 2001 From: arose26 <145766958+arose26@users.noreply.github.com> Date: Tue, 18 Aug 2026 09:27:21 -0400 Subject: [PATCH] feat: Add feature view versioning to Hazelcast online store Thread registry.enable_online_feature_view_versioning through _map_name, which now delegates to compute_table_id, so that with versioning enabled each feature view version gets its own Hazelcast map (test_project_driver_stats_v2) instead of all versions sharing one. Hazelcast already namespaced as {project}_{table.name}, so this is the same compute_table_id convention the Milvus and FAISS stores use and the map name is unchanged when versioning is disabled. The update() and teardown() SQL statements bind the resolved name to a local first rather than calling _map_name inside the f-string: line breaks inside an f-string replacement field are a Python 3.12 feature and this package supports 3.10. HazelcastOnlineStore is added to the versioned-read allowlist. Its online_read iterates the requested entity keys and appends (None, None) for a miss, so it returns one aligned entry per key as the OnlineStore contract requires; a test pins that property directly. Also adds Milvus to the VersionedOnlineReadNotSupported message, which has listed it as unsupported since it was allowlisted. Part of #2728. Closes #6174 Signed-off-by: arose26 <145766958+arose26@users.noreply.github.com> --- sdk/python/feast/errors.py | 2 +- .../hazelcast_online_store.py | 52 +++-- .../feast/infra/online_stores/online_store.py | 4 + .../online_store/test_hazelcast_versioning.py | 192 ++++++++++++++++++ 4 files changed, 236 insertions(+), 14 deletions(-) create mode 100644 sdk/python/tests/unit/infra/online_store/test_hazelcast_versioning.py diff --git a/sdk/python/feast/errors.py b/sdk/python/feast/errors.py index 515a6c39b11..02cd160a1f9 100644 --- a/sdk/python/feast/errors.py +++ b/sdk/python/feast/errors.py @@ -142,7 +142,7 @@ class VersionedOnlineReadNotSupported(FeastError): def __init__(self, store_name: str, version: int): super().__init__( f"Versioned feature reads (@v{version}) are not yet supported by {store_name}. " - f"Currently only SQLite, PostgreSQL, MySQL, FAISS, Redis, and DynamoDB support version-qualified feature references. " + f"Currently only SQLite, PostgreSQL, MySQL, FAISS, Redis, DynamoDB, Milvus, and Hazelcast support version-qualified feature references. " ) diff --git a/sdk/python/feast/infra/online_stores/hazelcast_online_store/hazelcast_online_store.py b/sdk/python/feast/infra/online_stores/hazelcast_online_store/hazelcast_online_store.py index 21359b45bca..d8418f0a4c3 100644 --- a/sdk/python/feast/infra/online_stores/hazelcast_online_store/hazelcast_online_store.py +++ b/sdk/python/feast/infra/online_stores/hazelcast_online_store/hazelcast_online_store.py @@ -30,6 +30,7 @@ from feast import Entity, FeatureView, RepoConfig from feast.infra.key_encoding_utils import serialize_entity_key +from feast.infra.online_stores.helpers import compute_table_id from feast.infra.online_stores.online_store import OnlineStore from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto @@ -157,7 +158,13 @@ def online_write_batch( ) client = self._get_client(online_store_config) - fv_map = client.get_map(_map_name(config.project, table)) + fv_map = client.get_map( + _map_name( + config.project, + table, + config.registry.enable_online_feature_view_versioning, + ) + ) for entity_key, values, event_ts, created_ts in data: entity_key_str = base64.b64encode( @@ -206,7 +213,13 @@ def online_read( client = self._get_client(online_store_config) entries: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = [] - fv_map = client.get_map(_map_name(config.project, table)) + fv_map = client.get_map( + _map_name( + config.project, + table, + config.registry.enable_online_feature_view_versioning, + ) + ) hz_keys = [] entity_keys_str = {} @@ -269,8 +282,11 @@ def update( project = config.project for table in tables_to_keep: + map_name = _map_name( + project, table, config.registry.enable_online_feature_view_versioning + ) client.sql.execute( - f"""CREATE OR REPLACE MAPPING {_map_name(project, table)} ( + f"""CREATE OR REPLACE MAPPING {map_name} ( __key VARCHAR, {D_ENTITY_KEY} VARCHAR, {D_FEATURE_NAME} VARCHAR, @@ -287,12 +303,13 @@ def update( ).result() for table in tables_to_delete: - client.sql.execute( - f"DELETE FROM {_map_name(config.project, table)}" - ).result() - client.sql.execute( - f"DROP MAPPING IF EXISTS {_map_name(config.project, table)}" - ).result() + map_name = _map_name( + config.project, + table, + config.registry.enable_online_feature_view_versioning, + ) + client.sql.execute(f"DELETE FROM {map_name}").result() + client.sql.execute(f"DROP MAPPING IF EXISTS {map_name}").result() def teardown( self, @@ -310,9 +327,18 @@ def teardown( project = config.project for table in tables: - client.sql.execute(f"DELETE FROM {_map_name(config.project, table)}") - client.sql.execute(f"DROP MAPPING IF EXISTS {_map_name(project, table)}") + map_name = _map_name( + project, table, config.registry.enable_online_feature_view_versioning + ) + client.sql.execute(f"DELETE FROM {map_name}") + client.sql.execute(f"DROP MAPPING IF EXISTS {map_name}") -def _map_name(project: str, table: FeatureView) -> str: - return f"{project}_{table.name}" +def _map_name(project: str, table: FeatureView, enable_versioning: bool = False) -> str: + """Hazelcast map backing a feature view. + + Delegates to ``compute_table_id`` so that, with versioning enabled, each feature + view version gets its own map (``{project}_{name}_v{N}``). Hazelcast already + namespaced as ``{project}_{name}``, so the unversioned name is unchanged. + """ + return compute_table_id(project, table, enable_versioning) diff --git a/sdk/python/feast/infra/online_stores/online_store.py b/sdk/python/feast/infra/online_stores/online_store.py index cdf06639fe0..c48f2b37e2f 100644 --- a/sdk/python/feast/infra/online_stores/online_store.py +++ b/sdk/python/feast/infra/online_stores/online_store.py @@ -315,6 +315,10 @@ def _is_versioned_read_supported(self) -> bool: "feast.infra.online_stores.milvus_online_store.milvus", "MilvusOnlineStore", ), + ( + "feast.infra.online_stores.hazelcast_online_store.hazelcast_online_store", + "HazelcastOnlineStore", + ), ): try: import importlib diff --git a/sdk/python/tests/unit/infra/online_store/test_hazelcast_versioning.py b/sdk/python/tests/unit/infra/online_store/test_hazelcast_versioning.py new file mode 100644 index 00000000000..7f435ff6abf --- /dev/null +++ b/sdk/python/tests/unit/infra/online_store/test_hazelcast_versioning.py @@ -0,0 +1,192 @@ +"""Unit tests for Hazelcast online store feature view versioning.""" + +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +import pytest + +from feast import Entity +from feast.feature_view import FeatureView +from feast.field import Field +from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto +from feast.protos.feast.types.Value_pb2 import Value as ValueProto +from feast.types import Float32 +from feast.value_type import ValueType + +pytest.importorskip("hazelcast") + +from feast.infra.online_stores.hazelcast_online_store.hazelcast_online_store import ( # noqa: E402 + HazelcastOnlineStore, + HazelcastOnlineStoreConfig, + _map_name, +) + + +def _make_feature_view(name="driver_stats", version_number=None, version_tag=None): + entity = Entity( + name="driver_id", join_keys=["driver_id"], value_type=ValueType.INT64 + ) + fv = FeatureView( + name=name, + entities=[entity], + ttl=timedelta(days=1), + schema=[Field(name="trips_today", dtype=Float32)], + ) + if version_number is not None: + fv.current_version_number = version_number + if version_tag is not None: + fv.projection.version_tag = version_tag + return fv + + +def _make_config(project="test_project", versioning=False): + config = MagicMock() + config.project = project + config.entity_key_serialization_version = 3 + config.online_store = HazelcastOnlineStoreConfig(type="hazelcast") + config.registry.enable_online_feature_view_versioning = versioning + return config + + +class TestMapName: + def test_no_versioning(self): + assert ( + _map_name("test_project", _make_feature_view()) + == "test_project_driver_stats" + ) + + def test_versioning_disabled_ignores_version(self): + fv = _make_feature_view(version_number=2) + assert _map_name("test_project", fv, False) == "test_project_driver_stats" + + def test_versioning_enabled_with_version(self): + fv = _make_feature_view(version_number=2) + assert _map_name("test_project", fv, True) == "test_project_driver_stats_v2" + + def test_projection_version_tag_takes_priority(self): + fv = _make_feature_view(version_number=1, version_tag=3) + assert _map_name("test_project", fv, True) == "test_project_driver_stats_v3" + + def test_version_zero_no_suffix(self): + fv = _make_feature_view(version_number=0) + assert _map_name("test_project", fv, True) == "test_project_driver_stats" + + def test_versions_do_not_collide(self): + v1 = _map_name("p", _make_feature_view(version_number=1), True) + v2 = _map_name("p", _make_feature_view(version_number=2), True) + assert v1 != v2 + + +@pytest.fixture +def store_with_mock_client(): + store = HazelcastOnlineStore() + client = MagicMock() + with patch.object(HazelcastOnlineStore, "_get_client", return_value=client): + yield store, client + + +class TestVersionedMapNamesReachHazelcast: + @pytest.mark.parametrize( + "versioning, expected", + [(False, "test_project_driver_stats"), (True, "test_project_driver_stats_v2")], + ) + def test_write_targets_the_versioned_map( + self, store_with_mock_client, versioning, expected + ): + store, client = store_with_mock_client + fv = _make_feature_view(version_number=2) + entity_key = EntityKeyProto( + join_keys=["driver_id"], entity_values=[ValueProto(int64_val=1)] + ) + + store.online_write_batch( + _make_config(versioning=versioning), + fv, + [ + ( + entity_key, + {"trips_today": ValueProto(float_val=1.0)}, + datetime(2024, 1, 1, tzinfo=timezone.utc), + None, + ) + ], + None, + ) + + client.get_map.assert_called_once_with(expected) + + def test_read_targets_the_versioned_map(self, store_with_mock_client): + store, client = store_with_mock_client + client.get_map.return_value.get_all.return_value.result.return_value = {} + fv = _make_feature_view(version_number=2) + entity_key = EntityKeyProto( + join_keys=["driver_id"], entity_values=[ValueProto(int64_val=1)] + ) + + store.online_read( + _make_config(versioning=True), fv, [entity_key], ["trips_today"] + ) + + client.get_map.assert_called_once_with("test_project_driver_stats_v2") + + def test_update_creates_the_versioned_mapping(self, store_with_mock_client): + store, client = store_with_mock_client + fv = _make_feature_view(version_number=2) + + store.update(_make_config(versioning=True), [], [fv], [], [], partial=False) + + sql = client.sql.execute.call_args[0][0] + assert "CREATE OR REPLACE MAPPING test_project_driver_stats_v2 (" in sql + + def test_update_drops_the_versioned_mapping(self, store_with_mock_client): + store, client = store_with_mock_client + fv = _make_feature_view(name="old_stats", version_number=1) + + store.update(_make_config(versioning=True), [fv], [], [], [], partial=False) + + statements = [c[0][0] for c in client.sql.execute.call_args_list] + assert "DELETE FROM test_project_old_stats_v1" in statements + assert "DROP MAPPING IF EXISTS test_project_old_stats_v1" in statements + + def test_teardown_drops_the_versioned_mapping(self, store_with_mock_client): + store, client = store_with_mock_client + fv = _make_feature_view(version_number=2) + + store.teardown(_make_config(versioning=True), [fv], []) + + statements = [c[0][0] for c in client.sql.execute.call_args_list] + assert "DELETE FROM test_project_driver_stats_v2" in statements + assert "DROP MAPPING IF EXISTS test_project_driver_stats_v2" in statements + + +class TestVersionedReadSupport: + def test_hazelcast_is_allowlisted(self): + """online_read yields one aligned entry per requested key, so this is honest.""" + store = HazelcastOnlineStore() + store._versioned_read_supported = None + assert store._is_versioned_read_supported() is True + + def test_versioned_ref_does_not_raise(self): + store = HazelcastOnlineStore() + store._versioned_read_supported = None + fv = _make_feature_view(version_tag=2) + store._check_versioned_read_support([(fv, ["trips_today"])]) + + def test_read_returns_one_entry_per_requested_key(self, store_with_mock_client): + """The property the allowlist actually depends on: misses yield (None, None).""" + store, client = store_with_mock_client + client.get_map.return_value.get_all.return_value.result.return_value = {} + fv = _make_feature_view(version_number=2) + entity_keys = [ + EntityKeyProto( + join_keys=["driver_id"], entity_values=[ValueProto(int64_val=i)] + ) + for i in (1, 2, 3) + ] + + result = store.online_read( + _make_config(versioning=True), fv, entity_keys, ["trips_today"] + ) + + assert len(result) == len(entity_keys) + assert all(entry == (None, None) for entry in result)