feast-dev
@@ -197,7 +197,7 @@ Versioning is supported on all three feature view types:
 ## Online Store Support
 
 {% hint style="info" %}
-**Currently, version-qualified online reads (`@v<N>`) are only supported with the SQLite online store.** Support for additional online stores (Redis, DynamoDB, Bigtable, Postgres, etc.) will be added based on community priority.
+**Version-qualified online reads (`@v<N>`) are supported on the SQLite, PostgreSQL, MySQL, Redis, DynamoDB, Milvus, FAISS and Cassandra online stores.** Support for the remaining online stores will be added based on community priority.
 
 If you need versioned online reads for a specific online store, please [open a GitHub issue](https://github.com/feast-dev/feast/issues/new) describing your use case and which store you need. This helps us prioritize development.
 {% endhint %}
@@ -222,7 +222,7 @@ ambiguity, the following characters are reserved and must not appear in feature
 
 ## Known Limitations
 
-- **Online store coverage** — Version-qualified reads (`@v<N>`) are SQLite-only today. Other online stores are follow-up work.
+- **Online store coverage** — Version-qualified reads (`@v<N>`) are supported on SQLite, PostgreSQL, MySQL, Redis, DynamoDB, Milvus, FAISS and Cassandra. The remaining online stores are follow-up work.
 - **Offline store versioning** — Versioned historical retrieval is not yet supported.
 - **Version deletion** — There is no mechanism to prune old versions from the registry.
 - **Cross-version joins** — Joining features from different versions of the same feature view in `get_historical_features` is not supported.

@@ -19,6 +19,7 @@
 """
 
 import logging
+import re
 from datetime import datetime
 from typing import (
     Any,
@@ -47,6 +48,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_versioned_name
 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
@@ -108,6 +110,10 @@
 
 DROP_TABLE_CQL_TEMPLATE = "DROP TABLE IF EXISTS {fqtable};"
 
+SELECT_KEYSPACE_TABLES_CQL_TEMPLATE = (
+    "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ?;"
+)
+
 # op_name -> (cql template string, prepare boolean)
 CQL_TEMPLATE_MAP = {
     # Queries/DML, statements to be prepared
@@ -116,6 +122,8 @@
     # DDL, do not prepare these
     "drop": (DROP_TABLE_CQL_TEMPLATE, False),
     "create": (CREATE_TABLE_CQL_TEMPLATE, False),
+    # Schema introspection, used to find every version of a feature view
+    "select_keyspace_tables": (SELECT_KEYSPACE_TABLES_CQL_TEMPLATE, True),
 }
 
 # Logger
@@ -686,11 +694,15 @@ def update(
             tables_to_keep: Tables to keep in the Online Store.
         """
         project = config.project
+        versioning = config.registry.enable_online_feature_view_versioning
 
         for table in tables_to_keep:
             self._create_table(config, project, table)
         for table in tables_to_delete:
-            self._drop_table(config, project, table)
+            if versioning:
+                self._drop_all_version_tables(config, project, table)
+            else:
+                self._drop_table(config, project, table)
 
     def teardown(
         self,
@@ -706,17 +718,31 @@ def teardown(
             tables: Tables to delete from the feature repo.
         """
         project = config.project
+        versioning = config.registry.enable_online_feature_view_versioning
 
         for table in tables:
-            self._drop_table(config, project, table)
+            if versioning:
+                self._drop_all_version_tables(config, project, table)
+            else:
+                self._drop_table(config, project, table)
 
     @staticmethod
-    def _fq_table_name(keyspace: str, project: str, table: FeatureView) -> str:
+    def _fq_table_name(
+        keyspace: str,
+        project: str,
+        table: FeatureView,
+        enable_versioning: bool = False,
+    ) -> str:
         """
         Generate a fully-qualified table name,
         including quotes and keyspace.
+
+        When feature view versioning is enabled, the version of the view is
+        appended to the table name (``driver_stats_v2``), so that each version
+        of a feature view is stored in a table of its own.
         """
-        return f'"{keyspace}"."{project}_{table.name}"'
+        versioned_name = compute_versioned_name(table, enable_versioning)
+        return f'"{keyspace}"."{project}_{versioned_name}"'
 
     def _write_rows_concurrently(
         self,
@@ -727,7 +753,12 @@ def _write_rows_concurrently(
     ):
         session: Session = self._get_session(config)
         keyspace: str = self._keyspace
-        fqtable = CassandraOnlineStore._fq_table_name(keyspace, project, table)
+        fqtable = CassandraOnlineStore._fq_table_name(
+            keyspace,
+            project,
+            table,
+            config.registry.enable_online_feature_view_versioning,
+        )
         insert_cql = self._get_cql_statement(config, "insert4", fqtable=fqtable)
         #
         execute_concurrent_with_args(
@@ -751,7 +782,12 @@ def _read_rows_by_entity_keys(
         """
         session: Session = self._get_session(config)
         keyspace: str = self._keyspace
-        fqtable = CassandraOnlineStore._fq_table_name(keyspace, project, table)
+        fqtable = CassandraOnlineStore._fq_table_name(
+            keyspace,
+            project,
+            table,
+            config.registry.enable_online_feature_view_versioning,
+        )
         projection_columns = "*" if columns is None else ", ".join(columns)
         select_cql = self._get_cql_statement(
             config,
@@ -789,16 +825,56 @@ def _drop_table(
         """Handle the CQL (low-level) deletion of a table."""
         session: Session = self._get_session(config)
         keyspace: str = self._keyspace
-        fqtable = CassandraOnlineStore._fq_table_name(keyspace, project, table)
+        fqtable = CassandraOnlineStore._fq_table_name(
+            keyspace,
+            project,
+            table,
+            config.registry.enable_online_feature_view_versioning,
+        )
         drop_cql = self._get_cql_statement(config, "drop", fqtable)
         logger.info(f"Deleting table {fqtable}.")
         session.execute(drop_cql)
 
+    def _drop_all_version_tables(
+        self,
+        config: RepoConfig,
+        project: str,
+        table: FeatureView,
+    ):
+        """
+        Handle the CQL (low-level) deletion of every version of a table.
+
+        The in-memory feature view carries a single version, but the keyspace
+        may hold one table per version ever written; dropping only the current
+        one would leave the rest behind. ``system_schema`` cannot match a
+        pattern, so the keyspace is listed and the names are filtered here.
+        """
+        session: Session = self._get_session(config)
+        keyspace: str = self._keyspace
+        base = f"{project}_{table.name}"
+        version_pattern = re.compile(rf"^{re.escape(base)}(_v[0-9]+)?$")
+        list_cql = self._get_cql_statement(config, "select_keyspace_tables", fqtable="")
+        table_names = [
+            row.table_name
+            for row in session.execute(list_cql, [keyspace])
+            if version_pattern.match(row.table_name)
+        ]
+        for table_name in table_names:
+            fqtable = f'"{keyspace}"."{table_name}"'
+            drop_cql = self._get_cql_statement(config, "drop", fqtable)
+            logger.info(f"Deleting table {fqtable}.")
+            session.execute(drop_cql)
+
     def _create_table(self, config: RepoConfig, project: str, table: FeatureView):
         """Handle the CQL (low-level) creation of a table."""
         session: Session = self._get_session(config)
         keyspace: str = self._keyspace
-        fqtable = CassandraOnlineStore._fq_table_name(keyspace, project, table)
+        fqtable = CassandraOnlineStore._fq_table_name(
+            keyspace,
+            project,
+            table,
+            config.registry.enable_online_feature_view_versioning,
+        )
         create_cql = self._get_cql_statement(config, "create", fqtable)
         logger.info(f"Creating table {fqtable}.")
         session.execute(create_cql)

@@ -315,6 +315,10 @@ def _is_versioned_read_supported(self) -> bool:
                 "feast.infra.online_stores.milvus_online_store.milvus",
                 "MilvusOnlineStore",
             ),
+            (
+                "feast.infra.online_stores.cassandra_online_store.cassandra_online_store",
+                "CassandraOnlineStore",
+            ),
         ):
             try:
                 import importlib

@@ -0,0 +1,189 @@
+"""Unit tests for Cassandra online store feature view versioning."""
+
+from datetime import timedelta
+from unittest.mock import MagicMock
+
+import pytest
+
+# Skip the entire module when the optional cassandra-driver is not installed.
+pytest.importorskip("cassandra", reason="cassandra-driver not installed")
+
+from feast import Entity, FeatureView  # noqa: E402
+from feast.field import Field  # noqa: E402
+from feast.infra.online_stores.cassandra_online_store.cassandra_online_store import (  # noqa: E402
+    CassandraOnlineStore,
+)
+from feast.types import Float32  # noqa: E402
+from feast.value_type import ValueType  # noqa: E402
+
+KEYSPACE = "feast_keyspace"
+PROJECT = "test_project"
+
+
+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=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 TestCassandraFqTableName:
+    """_fq_table_name appends the version only when versioning is enabled."""
+
+    def test_no_versioning(self):
+        fv = _make_feature_view()
+        assert (
+            CassandraOnlineStore._fq_table_name(KEYSPACE, PROJECT, fv, False)
+            == '"feast_keyspace"."test_project_driver_stats"'
+        )
+
+    def test_versioning_defaults_to_off(self):
+        fv = _make_feature_view(version_number=2)
+        assert (
+            CassandraOnlineStore._fq_table_name(KEYSPACE, PROJECT, fv)
+            == '"feast_keyspace"."test_project_driver_stats"'
+        )
+
+    def test_versioning_disabled_ignores_version(self):
+        fv = _make_feature_view(version_number=3)
+        assert (
+            CassandraOnlineStore._fq_table_name(KEYSPACE, PROJECT, fv, False)
+            == '"feast_keyspace"."test_project_driver_stats"'
+        )
+
+    def test_versioning_enabled_no_version_set(self):
+        fv = _make_feature_view()
+        assert (
+            CassandraOnlineStore._fq_table_name(KEYSPACE, PROJECT, fv, True)
+            == '"feast_keyspace"."test_project_driver_stats"'
+        )
+
+    def test_versioning_enabled_with_current_version_number(self):
+        fv = _make_feature_view(version_number=2)
+        assert (
+            CassandraOnlineStore._fq_table_name(KEYSPACE, PROJECT, fv, True)
+            == '"feast_keyspace"."test_project_driver_stats_v2"'
+        )
+
+    def test_projection_version_tag_takes_priority(self):
+        fv = _make_feature_view(version_number=1, version_tag=3)
+        assert (
+            CassandraOnlineStore._fq_table_name(KEYSPACE, PROJECT, fv, True)
+            == '"feast_keyspace"."test_project_driver_stats_v3"'
+        )
+
+    def test_version_zero_has_no_suffix(self):
+        fv = _make_feature_view(version_number=0)
+        assert (
+            CassandraOnlineStore._fq_table_name(KEYSPACE, PROJECT, fv, True)
+            == '"feast_keyspace"."test_project_driver_stats"'
+        )
+
+    def test_versions_get_separate_tables(self):
+        v1 = _make_feature_view(version_number=1)
+        v2 = _make_feature_view(version_number=2)
+        assert CassandraOnlineStore._fq_table_name(
+            KEYSPACE, PROJECT, v1, True
+        ) != CassandraOnlineStore._fq_table_name(KEYSPACE, PROJECT, v2, True)
+
+
+class TestCassandraVersionedReadSupport:
+    """A version-qualified read no longer raises on Cassandra."""
+
+    def test_allowed_with_version_tag(self):
+        store = CassandraOnlineStore()
+        fv = _make_feature_view()
+        fv.projection.version_tag = 2
+        # Should not raise VersionedOnlineReadNotSupported
+        store._check_versioned_read_support([(fv, ["trips_today"])])
+
+    def test_allowed_without_version_tag(self):
+        store = CassandraOnlineStore()
+        fv = _make_feature_view()
+        store._check_versioned_read_support([(fv, ["trips_today"])])
+
+
+class TestCassandraDropAllVersionTables:
+    """Deleting a versioned feature view removes every one of its tables."""
+
+    @staticmethod
+    def _store_with_tables(table_names):
+        store = CassandraOnlineStore()
+        store._keyspace = KEYSPACE
+        # The cache is a class attribute; give this instance one of its own.
+        store._prepared_statements = {}
+        session = MagicMock()
+        session.execute.return_value = [
+            MagicMock(table_name=name) for name in table_names
+        ]
+        store._get_session = MagicMock(return_value=session)
+        return store, session
+
+    def _dropped(self, session):
+        return [
+            call.args[0]
+            for call in session.execute.call_args_list
+            if isinstance(call.args[0], str) and call.args[0].startswith("DROP TABLE")
+        ]
+
+    def test_drops_base_and_every_version(self):
+        store, session = self._store_with_tables(
+            [
+                "test_project_driver_stats",
+                "test_project_driver_stats_v1",
+                "test_project_driver_stats_v2",
+            ]
+        )
+        store._drop_all_version_tables(
+            _make_config(versioning=True), PROJECT, _make_feature_view()
+        )
+        dropped = self._dropped(session)
+        assert len(dropped) == 3
+        assert any('"test_project_driver_stats_v2"' in cql for cql in dropped)
+        assert any('"test_project_driver_stats_v1"' in cql for cql in dropped)
+
+    def test_leaves_other_feature_views_alone(self):
+        store, session = self._store_with_tables(
+            [
+                "test_project_driver_stats",
+                "test_project_driver_stats_v1",
+                "test_project_driver_stats_extra",
+                "test_project_driver_stats_extra_v1",
+                "test_project_other_view_v1",
+            ]
+        )
+        store._drop_all_version_tables(
+            _make_config(versioning=True), PROJECT, _make_feature_view()
+        )
+        dropped = self._dropped(session)
+        assert len(dropped) == 2
+        assert not any("extra" in cql for cql in dropped)
+        assert not any("other_view" in cql for cql in dropped)
+
+    def test_teardown_without_versioning_drops_only_the_view(self):
+        store, session = self._store_with_tables([])
+        store.teardown(_make_config(versioning=False), [_make_feature_view()], [])
+        dropped = self._dropped(session)
+        assert dropped == [
+            'DROP TABLE IF EXISTS "feast_keyspace"."test_project_driver_stats";'
+        ]