diff --git a/docs/reference/alpha-feature-view-versioning.md b/docs/reference/alpha-feature-view-versioning.md index 5cdf2845ebc..e2b8f40d147 100644 --- a/docs/reference/alpha-feature-view-versioning.md +++ b/docs/reference/alpha-feature-view-versioning.md @@ -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`) 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`) 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`) are SQLite-only today. Other online stores are follow-up work. +- **Online store coverage** — Version-qualified reads (`@v`) 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. diff --git a/sdk/python/feast/infra/online_stores/cassandra_online_store/cassandra_online_store.py b/sdk/python/feast/infra/online_stores/cassandra_online_store/cassandra_online_store.py index cea3ed4f87d..b68c16f6a1c 100644 --- a/sdk/python/feast/infra/online_stores/cassandra_online_store/cassandra_online_store.py +++ b/sdk/python/feast/infra/online_stores/cassandra_online_store/cassandra_online_store.py @@ -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) diff --git a/sdk/python/feast/infra/online_stores/online_store.py b/sdk/python/feast/infra/online_stores/online_store.py index cdf06639fe0..4eca403abc8 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.cassandra_online_store.cassandra_online_store", + "CassandraOnlineStore", + ), ): try: import importlib diff --git a/sdk/python/tests/unit/infra/online_store/test_cassandra_versioning.py b/sdk/python/tests/unit/infra/online_store/test_cassandra_versioning.py new file mode 100644 index 00000000000..962af49386a --- /dev/null +++ b/sdk/python/tests/unit/infra/online_store/test_cassandra_versioning.py @@ -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";' + ]