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>
arose26 · Aug 18, 2026 · Aug 23, 2026 · Aug 18, 2026 · cd5bdc683d098278dbd0a591dc9dad686afa35d3
commit cd5bdc683d098278dbd0a591dc9dad686afa35d3
@@ -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)}"
@@ -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)