From cd5bdc683d098278dbd0a591dc9dad686afa35d3 Mon Sep 17 00:00:00 2001 From: arose26 <145766958+arose26@users.noreply.github.com> Date: Tue, 18 Aug 2026 07:49:20 -0400 Subject: [PATCH] feat: Add feature view versioning to HBase online store Thread registry.enable_online_feature_view_versioning through _table_id so that, when it is enabled, each feature view version gets its own HBase table (test_project:driver_stats_v2) instead of all versions sharing one. The version is appended via compute_versioned_name rather than by calling compute_table_id, because HBase addresses tables as namespace:table while compute_table_id joins with an underscore. Building the name here keeps the ':' separator, and with versioning disabled the table name is unchanged. Row keys stay unversioned. They are already suffixed with the feature view name to disambiguate views sharing a table, and since the table itself is now version-scoped the version would be redundant there; leaving them alone keeps existing keys stable and keeps the write and read paths computing identical keys, which a test pins. HbaseOnlineStore is deliberately not added to the versioned-read allowlist: online_read builds its result by iterating the rows HBase returned rather than the entity keys requested, so a missing row shortens the list instead of yielding (None, None) in place. That contract gap is pre-existing and independent of versioning. See the PR description. Part of #2728. Closes #6175 Signed-off-by: arose26 <145766958+arose26@users.noreply.github.com> --- .../online_stores/hbase_online_store/hbase.py | 30 +- .../online_store/test_hbase_versioning.py | 265 ++++++++++++++++++ 2 files changed, 287 insertions(+), 8 deletions(-) create mode 100644 sdk/python/tests/unit/infra/online_store/test_hbase_versioning.py diff --git a/sdk/python/feast/infra/online_stores/hbase_online_store/hbase.py b/sdk/python/feast/infra/online_stores/hbase_online_store/hbase.py index dc48d2c4efc..bf8688bed7f 100644 --- a/sdk/python/feast/infra/online_stores/hbase_online_store/hbase.py +++ b/sdk/python/feast/infra/online_stores/hbase_online_store/hbase.py @@ -9,7 +9,7 @@ from feast import Entity from feast.feature_view import FeatureView -from feast.infra.online_stores.helpers import compute_entity_id +from feast.infra.online_stores.helpers import compute_entity_id, compute_versioned_name from feast.infra.online_stores.online_store import OnlineStore from feast.infra.utils.hbase_utils import HBaseConnector, HbaseConstants from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto @@ -94,7 +94,9 @@ def online_write_batch( hbase = HBaseConnector(self._get_conn(config)) project = config.project - table_name = self._table_id(project, table) + table_name = self._table_id( + project, table, config.registry.enable_online_feature_view_versioning + ) b = hbase.batch(table_name) for entity_key, values, timestamp, created_ts in data: @@ -145,7 +147,9 @@ def online_read( """ hbase = HBaseConnector(self._get_conn(config)) project = config.project - table_name = self._table_id(project, table) + table_name = self._table_id( + project, table, config.registry.enable_online_feature_view_versioning + ) result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = [] @@ -199,12 +203,16 @@ def update( # We don't create any special state for the entites in this implementation. for table in tables_to_keep: - table_name = self._table_id(project, table) + table_name = self._table_id( + project, table, config.registry.enable_online_feature_view_versioning + ) if not hbase.check_if_table_exist(table_name): hbase.create_table_with_default_cf(table_name) for table in tables_to_delete: - table_name = self._table_id(project, table) + table_name = self._table_id( + project, table, config.registry.enable_online_feature_view_versioning + ) hbase.delete_table(table_name) def teardown( @@ -224,7 +232,9 @@ def teardown( project = config.project for table in tables: - table_name = self._table_id(project, table) + table_name = self._table_id( + project, table, config.registry.enable_online_feature_view_versioning + ) hbase.delete_table(table_name) def _hbase_row_key( @@ -255,12 +265,16 @@ def _hbase_row_key( # colocated. return f"{entity_id}#{feature_view_name}".encode() - def _table_id(self, project: str, table: FeatureView) -> str: + def _table_id( + self, project: str, table: FeatureView, enable_versioning: bool = False + ) -> str: """ Returns table name given the project_name and the feature_view. Args: project: Name of the feast project. table: Feast FeatureView. + enable_versioning: When set, the feature view's version is appended to + the table name so each version gets its own HBase table. """ - return f"{project}:{table.name}" + return f"{project}:{compute_versioned_name(table, enable_versioning)}" diff --git a/sdk/python/tests/unit/infra/online_store/test_hbase_versioning.py b/sdk/python/tests/unit/infra/online_store/test_hbase_versioning.py new file mode 100644 index 00000000000..fb2fdfe24d3 --- /dev/null +++ b/sdk/python/tests/unit/infra/online_store/test_hbase_versioning.py @@ -0,0 +1,265 @@ +"""Unit tests for HBase online store feature view versioning.""" + +import struct +from datetime import datetime, timedelta +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("happybase") + +from feast.infra.online_stores.hbase_online_store.hbase import ( # noqa: E402 + HbaseOnlineStore, +) + + +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.registry.enable_online_feature_view_versioning = versioning + return config + + +class TestTableId: + """HBase names tables `namespace:table`, so the version belongs on the table half.""" + + def test_no_versioning(self): + store = HbaseOnlineStore() + assert store._table_id("test_project", _make_feature_view()) == ( + "test_project:driver_stats" + ) + + def test_versioning_disabled_ignores_version(self): + store = HbaseOnlineStore() + fv = _make_feature_view(version_number=2) + assert store._table_id("test_project", fv, enable_versioning=False) == ( + "test_project:driver_stats" + ) + + def test_versioning_enabled_with_version(self): + store = HbaseOnlineStore() + fv = _make_feature_view(version_number=2) + assert store._table_id("test_project", fv, enable_versioning=True) == ( + "test_project:driver_stats_v2" + ) + + def test_projection_version_tag_takes_priority(self): + store = HbaseOnlineStore() + fv = _make_feature_view(version_number=1, version_tag=3) + assert store._table_id("test_project", fv, enable_versioning=True) == ( + "test_project:driver_stats_v3" + ) + + def test_version_zero_no_suffix(self): + store = HbaseOnlineStore() + fv = _make_feature_view(version_number=0) + assert store._table_id("test_project", fv, enable_versioning=True) == ( + "test_project:driver_stats" + ) + + def test_namespace_separator_is_preserved(self): + """`compute_table_id` would join with `_`; HBase requires `:`.""" + store = HbaseOnlineStore() + fv = _make_feature_view(version_number=2) + assert store._table_id("p", fv, enable_versioning=True).startswith("p:") + + +@pytest.fixture +def store_with_mock_connector(): + store = HbaseOnlineStore() + with ( + patch.object(HbaseOnlineStore, "_get_conn"), + patch( + "feast.infra.online_stores.hbase_online_store.hbase.HBaseConnector" + ) as connector_cls, + ): + yield store, connector_cls.return_value + + +class TestVersionedTableNamesReachHBase: + """The resolved name must be the one the store actually operates on.""" + + @pytest.mark.parametrize( + "versioning, expected", + [(False, "test_project:driver_stats"), (True, "test_project:driver_stats_v2")], + ) + def test_write_targets_the_versioned_table( + self, store_with_mock_connector, versioning, expected + ): + store, connector = store_with_mock_connector + config = _make_config(versioning=versioning) + fv = _make_feature_view(version_number=2) + entity_key = EntityKeyProto( + join_keys=["driver_id"], entity_values=[ValueProto(int64_val=1)] + ) + + store.online_write_batch( + config, + fv, + [ + ( + entity_key, + {"trips_today": ValueProto(float_val=1.0)}, + datetime(2024, 1, 1), + None, + ) + ], + None, + ) + + connector.batch.assert_called_once_with(expected) + + def test_read_targets_the_versioned_table(self, store_with_mock_connector): + store, connector = store_with_mock_connector + connector.rows.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"] + ) + + assert connector.rows.call_args[0][0] == "test_project:driver_stats_v2" + + def test_update_creates_and_deletes_versioned_tables( + self, store_with_mock_connector + ): + store, connector = store_with_mock_connector + connector.check_if_table_exist.return_value = False + keep, delete = ( + _make_feature_view(version_number=2), + _make_feature_view(name="old_stats", version_number=1), + ) + + store.update( + _make_config(versioning=True), [delete], [keep], [], [], partial=False + ) + + connector.create_table_with_default_cf.assert_called_once_with( + "test_project:driver_stats_v2" + ) + connector.delete_table.assert_called_once_with("test_project:old_stats_v1") + + def test_teardown_deletes_the_versioned_table(self, store_with_mock_connector): + store, connector = store_with_mock_connector + fv = _make_feature_view(version_number=2) + + store.teardown(_make_config(versioning=True), [fv], []) + + connector.delete_table.assert_called_once_with("test_project:driver_stats_v2") + + def test_row_keys_stay_unversioned(self, store_with_mock_connector): + """The table is the namespace, so row keys need no version and stay stable.""" + store, connector = store_with_mock_connector + connector.rows.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"] + ) + + row_keys = connector.rows.call_args[1]["row_keys"] + assert all(b"_v2" not in key for key in row_keys) + assert all(key.endswith(b"#driver_stats") for key in row_keys) + + def test_write_and_read_agree_on_row_keys(self, store_with_mock_connector): + """Versioning must not desynchronise the two paths.""" + store, connector = store_with_mock_connector + connector.rows.return_value = [] + config = _make_config(versioning=True) + fv = _make_feature_view(version_number=2) + entity_key = EntityKeyProto( + join_keys=["driver_id"], entity_values=[ValueProto(int64_val=1)] + ) + + store.online_write_batch( + config, + fv, + [ + ( + entity_key, + {"trips_today": ValueProto(float_val=1.0)}, + datetime(2024, 1, 1), + None, + ) + ], + None, + ) + written_key = connector.batch.return_value.put.call_args[0][0] + + store.online_read(config, fv, [entity_key], ["trips_today"]) + read_key = connector.rows.call_args[1]["row_keys"][0] + + assert written_key == read_key + + +class TestVersionsAreIsolated: + def test_two_versions_use_different_tables(self, store_with_mock_connector): + store, connector = store_with_mock_connector + config = _make_config(versioning=True) + v1, v2 = ( + _make_feature_view(version_number=1), + _make_feature_view(version_number=2), + ) + + assert store._table_id(config.project, v1, True) != store._table_id( + config.project, v2, True + ) + + def test_event_ts_still_round_trips(self, store_with_mock_connector): + """Guard the read decode path while the table name changes underneath it.""" + store, connector = store_with_mock_connector + ts = datetime(2024, 1, 1, 12, 0, 0) + packed = struct.pack(">L", int(ts.timestamp())) + value = ValueProto(float_val=2.0) + connector.rows.return_value = [ + ( + b"key#driver_stats", + { + b"data:trips_today": value.SerializeToString(), + b"data:event_ts": packed, + }, + ) + ] + fv = _make_feature_view(version_number=2) + entity_key = EntityKeyProto( + join_keys=["driver_id"], entity_values=[ValueProto(int64_val=1)] + ) + + result = store.online_read( + _make_config(versioning=True), fv, [entity_key], ["trips_today"] + ) + + assert len(result) == 1 + assert result[0][1]["trips_today"].float_val == pytest.approx(2.0)