|
| 1 | +"""Integration tests for DynamoDB online store feature view versioning. |
| 2 | +
|
| 3 | +Run with: pytest --integration sdk/python/tests/integration/online_store/test_dynamodb_versioning.py |
| 4 | +
|
| 5 | +Uses moto to mock the DynamoDB service (no Docker required). |
| 6 | +""" |
| 7 | + |
| 8 | +import os |
| 9 | +from datetime import datetime, timedelta, timezone |
| 10 | + |
| 11 | +import pytest |
| 12 | + |
| 13 | +from feast import Entity, FeatureView |
| 14 | +from feast.field import Field |
| 15 | +from feast.infra.online_stores.dynamodb import DynamoDBOnlineStore |
| 16 | +from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto |
| 17 | +from feast.protos.feast.types.Value_pb2 import Value as ValueProto |
| 18 | +from feast.repo_config import RegistryConfig, RepoConfig |
| 19 | +from feast.types import Float32, Int64 |
| 20 | +from feast.value_type import ValueType |
| 21 | + |
| 22 | + |
| 23 | +def _make_feature_view(name="driver_stats", version="latest"): |
| 24 | + entity = Entity( |
| 25 | + name="driver_id", |
| 26 | + join_keys=["driver_id"], |
| 27 | + value_type=ValueType.INT64, |
| 28 | + ) |
| 29 | + return FeatureView( |
| 30 | + name=name, |
| 31 | + entities=[entity], |
| 32 | + ttl=timedelta(days=1), |
| 33 | + schema=[ |
| 34 | + Field(name="driver_id", dtype=Int64), |
| 35 | + Field(name="trips_today", dtype=Int64), |
| 36 | + Field(name="avg_rating", dtype=Float32), |
| 37 | + ], |
| 38 | + version=version, |
| 39 | + ) |
| 40 | + |
| 41 | + |
| 42 | +def _make_entity_key(driver_id: int) -> EntityKeyProto: |
| 43 | + entity_key = EntityKeyProto() |
| 44 | + entity_key.join_keys.append("driver_id") |
| 45 | + val = ValueProto() |
| 46 | + val.int64_val = driver_id |
| 47 | + entity_key.entity_values.append(val) |
| 48 | + return entity_key |
| 49 | + |
| 50 | + |
| 51 | +def _write_and_read(store, config, fv, driver_id=1001, trips=42): |
| 52 | + entity_key = _make_entity_key(driver_id) |
| 53 | + val = ValueProto() |
| 54 | + val.int64_val = trips |
| 55 | + now = datetime.now(tz=timezone.utc) |
| 56 | + store.online_write_batch( |
| 57 | + config, fv, [(entity_key, {"trips_today": val}, now, now)], None |
| 58 | + ) |
| 59 | + return store.online_read(config, fv, [entity_key], ["trips_today"]) |
| 60 | + |
| 61 | + |
| 62 | +def _make_config(enable_versioning=False): |
| 63 | + from feast.infra.online_stores.dynamodb import DynamoDBOnlineStoreConfig |
| 64 | + |
| 65 | + return RepoConfig( |
| 66 | + project="test_project", |
| 67 | + provider="local", |
| 68 | + online_store=DynamoDBOnlineStoreConfig( |
| 69 | + type="dynamodb", |
| 70 | + region="us-east-1", |
| 71 | + ), |
| 72 | + registry=RegistryConfig( |
| 73 | + path="/tmp/test_dynamodb_registry.pb", |
| 74 | + enable_online_feature_view_versioning=enable_versioning, |
| 75 | + ), |
| 76 | + entity_key_serialization_version=3, |
| 77 | + ) |
| 78 | + |
| 79 | + |
| 80 | +@pytest.mark.integration |
| 81 | +class TestDynamoDBVersioningIntegration: |
| 82 | + """Integration tests for DynamoDB versioning using moto mock.""" |
| 83 | + |
| 84 | + @pytest.fixture(autouse=True) |
| 85 | + def setup_dynamodb(self): |
| 86 | + try: |
| 87 | + from moto import mock_dynamodb |
| 88 | + except ImportError: |
| 89 | + pytest.skip("moto not installed") |
| 90 | + |
| 91 | + # Set dummy AWS credentials for moto |
| 92 | + os.environ["AWS_ACCESS_KEY_ID"] = "testing" |
| 93 | + os.environ["AWS_SECRET_ACCESS_KEY"] = "testing" # noqa: S105 # pragma: allowlist secret |
| 94 | + os.environ["AWS_SECURITY_TOKEN"] = "testing" # noqa: S105 # pragma: allowlist secret |
| 95 | + os.environ["AWS_SESSION_TOKEN"] = "testing" # noqa: S105 # pragma: allowlist secret |
| 96 | + os.environ["AWS_DEFAULT_REGION"] = "us-east-1" |
| 97 | + |
| 98 | + with mock_dynamodb(): |
| 99 | + yield |
| 100 | + |
| 101 | + def test_write_read_without_versioning(self): |
| 102 | + config = _make_config(enable_versioning=False) |
| 103 | + store = DynamoDBOnlineStore() |
| 104 | + fv = _make_feature_view() |
| 105 | + store.update(config, [], [fv], [], [], False) |
| 106 | + |
| 107 | + result = _write_and_read(store, config, fv) |
| 108 | + assert result[0][1] is not None |
| 109 | + assert result[0][1]["trips_today"].int64_val == 42 |
| 110 | + |
| 111 | + def test_write_read_with_versioning_v1(self): |
| 112 | + config = _make_config(enable_versioning=True) |
| 113 | + store = DynamoDBOnlineStore() |
| 114 | + fv = _make_feature_view() |
| 115 | + fv.current_version_number = 1 |
| 116 | + store.update(config, [], [fv], [], [], False) |
| 117 | + |
| 118 | + result = _write_and_read(store, config, fv) |
| 119 | + assert result[0][1] is not None |
| 120 | + assert result[0][1]["trips_today"].int64_val == 42 |
| 121 | + |
| 122 | + def test_version_isolation(self): |
| 123 | + """Data written to v1 is not visible from v2.""" |
| 124 | + config = _make_config(enable_versioning=True) |
| 125 | + store = DynamoDBOnlineStore() |
| 126 | + |
| 127 | + fv_v1 = _make_feature_view() |
| 128 | + fv_v1.current_version_number = 1 |
| 129 | + store.update(config, [], [fv_v1], [], [], False) |
| 130 | + _write_and_read(store, config, fv_v1, driver_id=1001, trips=10) |
| 131 | + |
| 132 | + fv_v2 = _make_feature_view() |
| 133 | + fv_v2.current_version_number = 2 |
| 134 | + store.update(config, [], [fv_v2], [], [], False) |
| 135 | + |
| 136 | + entity_key = _make_entity_key(1001) |
| 137 | + result = store.online_read(config, fv_v2, [entity_key], ["trips_today"]) |
| 138 | + assert result[0] == (None, None) |
| 139 | + |
| 140 | + result = store.online_read(config, fv_v1, [entity_key], ["trips_today"]) |
| 141 | + assert result[0][1] is not None |
| 142 | + assert result[0][1]["trips_today"].int64_val == 10 |
| 143 | + |
| 144 | + def test_projection_version_tag_routes_to_correct_table(self): |
| 145 | + """projection.version_tag routes reads to the correct versioned DynamoDB table.""" |
| 146 | + config = _make_config(enable_versioning=True) |
| 147 | + store = DynamoDBOnlineStore() |
| 148 | + |
| 149 | + fv_v1 = _make_feature_view() |
| 150 | + fv_v1.current_version_number = 1 |
| 151 | + store.update(config, [], [fv_v1], [], [], False) |
| 152 | + _write_and_read(store, config, fv_v1, driver_id=1001, trips=100) |
| 153 | + |
| 154 | + fv_v2 = _make_feature_view() |
| 155 | + fv_v2.current_version_number = 2 |
| 156 | + store.update(config, [], [fv_v2], [], [], False) |
| 157 | + _write_and_read(store, config, fv_v2, driver_id=1001, trips=200) |
| 158 | + |
| 159 | + fv_read = _make_feature_view() |
| 160 | + fv_read.projection.version_tag = 1 |
| 161 | + entity_key = _make_entity_key(1001) |
| 162 | + result = store.online_read(config, fv_read, [entity_key], ["trips_today"]) |
| 163 | + assert result[0][1]["trips_today"].int64_val == 100 |
| 164 | + |
| 165 | + fv_read2 = _make_feature_view() |
| 166 | + fv_read2.projection.version_tag = 2 |
| 167 | + result = store.online_read(config, fv_read2, [entity_key], ["trips_today"]) |
| 168 | + assert result[0][1]["trips_today"].int64_val == 200 |
| 169 | + |
| 170 | + def test_teardown_versioned_table(self): |
| 171 | + """teardown() drops the versioned DynamoDB table without error.""" |
| 172 | + config = _make_config(enable_versioning=True) |
| 173 | + store = DynamoDBOnlineStore() |
| 174 | + |
| 175 | + fv = _make_feature_view() |
| 176 | + fv.current_version_number = 1 |
| 177 | + store.update(config, [], [fv], [], [], False) |
| 178 | + _write_and_read(store, config, fv) |
| 179 | + |
| 180 | + # Should not raise |
| 181 | + store.teardown(config, [fv], []) |
| 182 | + |
| 183 | + def test_update_deletes_versioned_table(self): |
| 184 | + """update() with tables_to_delete correctly drops versioned DynamoDB tables.""" |
| 185 | + config = _make_config(enable_versioning=True) |
| 186 | + store = DynamoDBOnlineStore() |
| 187 | + |
| 188 | + fv = _make_feature_view() |
| 189 | + fv.current_version_number = 1 |
| 190 | + store.update(config, [], [fv], [], [], False) |
| 191 | + _write_and_read(store, config, fv, driver_id=1001, trips=50) |
| 192 | + |
| 193 | + # Delete the versioned table |
| 194 | + store.update(config, [fv], [], [], [], False) |
0 commit comments