From 377eb992ee7b5b8651feff3c63ede8a5ff47ebd4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20S=C3=A1nchez?= Date: Wed, 29 Jul 2026 14:50:19 +0200 Subject: [PATCH] feat: add plan() support to DynamoDBOnlineStore MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DynamoDBOnlineStore has no plan() override, so it inherits OnlineStore.plan()'s no-op default: `feast plan` never reports any DynamoDB infrastructure changes, unlike SqliteOnlineStore and MilvusOnlineStore which both implement it. Adds DynamoDBTable (InfraObject) and DynamoDBOnlineStore.plan(), using the InfraObject proto's CustomInfra field (protos/feast/core/InfraObject.proto), which exists specifically so online stores can add InfraObject support without changes to the core proto -- no other in-tree store uses it yet. Note this only affects `feast plan`'s reporting: `feast apply`'s diff-based path (FeatureStore._should_use_plan(), which would call InfraObject.update()/teardown()) is gated to the local/sqlite provider only, so DynamoDBOnlineStore.update()/teardown() -- which already perform the real table creation/deletion -- are unaffected. Uses the corrected feature-view-list pattern (see #6658 / #6659): FeatureView.from_proto() and StreamFeatureView.from_proto() applied to their respective proto lists, not one applied to both. Signed-off-by: Carlos Sánchez --- .../feast/infra/online_stores/dynamodb.py | 92 +++++++++++ .../infra/online_store/test_dynamodb_plan.py | 153 ++++++++++++++++++ 2 files changed, 245 insertions(+) create mode 100644 sdk/python/tests/unit/infra/online_store/test_dynamodb_plan.py diff --git a/sdk/python/feast/infra/online_stores/dynamodb.py b/sdk/python/feast/infra/online_stores/dynamodb.py index 8c940fdc41b..b67c8a81f6b 100644 --- a/sdk/python/feast/infra/online_stores/dynamodb.py +++ b/sdk/python/feast/infra/online_stores/dynamodb.py @@ -14,6 +14,7 @@ import asyncio import contextlib import itertools +import json import logging from collections import OrderedDict, defaultdict from concurrent.futures import ThreadPoolExecutor @@ -24,13 +25,17 @@ from pydantic import StrictBool, StrictStr from feast import Entity, FeatureView, utils +from feast.infra.infra_object import InfraObject 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.supported_async_methods import SupportedAsyncMethods from feast.infra.utils.aws_utils import dynamo_write_items_async +from feast.protos.feast.core.InfraObject_pb2 import InfraObject as InfraObjectProto +from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto from feast.repo_config import FeastConfigBaseModel, RepoConfig +from feast.stream_feature_view import StreamFeatureView from feast.utils import get_user_agent try: @@ -379,6 +384,31 @@ def update( _get_table_name(online_config, config, table_to_delete), ) + def plan( + self, config: RepoConfig, desired_registry_proto: RegistryProto + ) -> List[InfraObject]: + online_config = config.online_store + assert isinstance(online_config, DynamoDBOnlineStoreConfig) + + # feature_views and stream_feature_views are distinct proto types + # (FeatureViewProto vs StreamFeatureViewProto); each needs its + # matching from_proto(), not one applied to both indiscriminately. + views = [ + FeatureView.from_proto(view) + for view in desired_registry_proto.feature_views + ] + [ + StreamFeatureView.from_proto(view) + for view in desired_registry_proto.stream_feature_views + ] + return [ + DynamoDBTable( + name=_get_table_name(online_config, config, view), + region=online_config.region, + endpoint_url=online_config.endpoint_url, + ) + for view in views + ] + def teardown( self, config: RepoConfig, @@ -1222,6 +1252,68 @@ def _get_table_name( ) +DYNAMODB_INFRA_OBJECT_CLASS_TYPE = "feast.infra.online_stores.dynamodb.DynamoDBTable" + + +class DynamoDBTable(InfraObject): + """ + A DynamoDB table managed by Feast, reported by DynamoDBOnlineStore.plan() + so `feast plan` can show DynamoDB table changes. + + Uses the InfraObject proto's CustomInfra field rather than a dedicated + proto message, since that field exists precisely to let online stores + add InfraObject support without changing the core InfraObject proto. + + Note: `feast apply` does not call update()/teardown() on this object -- + DynamoDBOnlineStore.update()/teardown() perform the actual table + creation and deletion directly (see FeatureStore._should_use_plan(), + which gates the new diff-based apply path to the local/sqlite provider + only). update()/teardown() here are no-ops that satisfy InfraObject's + abstract interface. + """ + + def __init__(self, name: str, region: str, endpoint_url: Optional[str] = None): + super().__init__(name) + self.region = region + self.endpoint_url = endpoint_url + + def to_infra_object_proto(self) -> InfraObjectProto: + return InfraObjectProto( + infra_object_class_type=DYNAMODB_INFRA_OBJECT_CLASS_TYPE, + custom_infra=InfraObjectProto.CustomInfra(field=self.to_proto()), + ) + + def to_proto(self) -> bytes: + return json.dumps( + { + "name": self.name, + "region": self.region, + "endpoint_url": self.endpoint_url, + } + ).encode("utf-8") + + @staticmethod + def from_infra_object_proto( + infra_object_proto: InfraObjectProto, + ) -> "DynamoDBTable": + return DynamoDBTable.from_proto(infra_object_proto.custom_infra.field) + + @staticmethod + def from_proto(serialized: bytes) -> "DynamoDBTable": + payload = json.loads(serialized.decode("utf-8")) + return DynamoDBTable( + name=payload["name"], + region=payload["region"], + endpoint_url=payload.get("endpoint_url"), + ) + + def update(self) -> None: + pass + + def teardown(self) -> None: + pass + + def _delete_table_idempotent( dynamodb_resource, table_name: str, diff --git a/sdk/python/tests/unit/infra/online_store/test_dynamodb_plan.py b/sdk/python/tests/unit/infra/online_store/test_dynamodb_plan.py new file mode 100644 index 00000000000..bb65a2adc5c --- /dev/null +++ b/sdk/python/tests/unit/infra/online_store/test_dynamodb_plan.py @@ -0,0 +1,153 @@ +from datetime import timedelta + +from feast.data_format import AvroFormat +from feast.data_source import KafkaSource +from feast.feature_view import FeatureView +from feast.field import Field +from feast.infra.infra_object import Infra +from feast.infra.offline_stores.dask import DaskOfflineStoreConfig +from feast.infra.offline_stores.file_source import FileSource +from feast.infra.online_stores.dynamodb import ( + DynamoDBOnlineStore, + DynamoDBOnlineStoreConfig, + DynamoDBTable, +) +from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto +from feast.repo_config import RepoConfig +from feast.stream_feature_view import StreamFeatureView +from feast.types import String + +REGISTRY = "s3://test_registry/registry.db" +PROJECT = "test_aws" +PROVIDER = "aws" +REGION = "us-west-2" + + +def _repo_config() -> RepoConfig: + return RepoConfig( + registry=REGISTRY, + project=PROJECT, + provider=PROVIDER, + online_store=DynamoDBOnlineStoreConfig(region=REGION), + offline_store=DaskOfflineStoreConfig(), + entity_key_serialization_version=3, + ) + + +def _feature_view(name: str) -> FeatureView: + return FeatureView( + name=name, + entities=[], + schema=[Field(name="value", dtype=String)], + ttl=timedelta(days=1), + online=True, + source=FileSource(path="dummy.parquet", timestamp_field="event_timestamp"), + ) + + +def _stream_feature_view(name: str) -> StreamFeatureView: + return StreamFeatureView( + name=name, + entities=[], + schema=[Field(name="value", dtype=String)], + source=KafkaSource( + name="dummy_kafka", + timestamp_field="event_timestamp", + message_format=AvroFormat(""), + kafka_bootstrap_servers="localhost:9092", + topic="dummy_topic", + batch_source=FileSource( + path="dummy.parquet", timestamp_field="event_timestamp" + ), + ), + ) + + +class TestDynamoDBOnlineStorePlan: + """DynamoDBOnlineStore previously had no plan() override at all, so + `feast plan` reported no DynamoDB infrastructure changes. This adds one, + following the InfraObject.CustomInfra extension point (see + protos/feast/core/InfraObject.proto) used by no other in-tree store yet.""" + + def test_plan_returns_one_table_per_feature_view(self): + config = _repo_config() + registry_proto = RegistryProto() + registry_proto.feature_views.append(_feature_view("view_a").to_proto()) + registry_proto.feature_views.append(_feature_view("view_b").to_proto()) + + infra_objects = DynamoDBOnlineStore().plan(config, registry_proto) + + assert sorted(o.name for o in infra_objects) == [ + f"{PROJECT}.view_a", + f"{PROJECT}.view_b", + ] + assert all(isinstance(o, DynamoDBTable) for o in infra_objects) + assert all(o.region == REGION for o in infra_objects) + + def test_plan_includes_stream_feature_views(self): + """Regression guard: FeatureView.from_proto() is @typechecked and + only accepts a FeatureViewProto, so it can't be applied to + stream_feature_views (StreamFeatureViewProto) too -- each list needs + its matching class (see the related fix in sqlite.py's plan()).""" + config = _repo_config() + registry_proto = RegistryProto() + registry_proto.stream_feature_views.append( + _stream_feature_view("driver_dropoffs_stream").to_proto() + ) + + infra_objects = DynamoDBOnlineStore().plan(config, registry_proto) + + assert [o.name for o in infra_objects] == [f"{PROJECT}.driver_dropoffs_stream"] + + def test_plan_includes_batch_and_stream_feature_views_together(self): + config = _repo_config() + registry_proto = RegistryProto() + registry_proto.feature_views.append(_feature_view("batch_view").to_proto()) + registry_proto.stream_feature_views.append( + _stream_feature_view("driver_dropoffs_stream").to_proto() + ) + + infra_objects = DynamoDBOnlineStore().plan(config, registry_proto) + + assert sorted(o.name for o in infra_objects) == [ + f"{PROJECT}.batch_view", + f"{PROJECT}.driver_dropoffs_stream", + ] + + def test_plan_empty_registry_produces_empty_plan(self): + config = _repo_config() + registry_proto = RegistryProto() + + assert DynamoDBOnlineStore().plan(config, registry_proto) == [] + + +class TestDynamoDBTableProtoRoundTrip: + """This is the mechanism `feast plan` relies on: Infra gets serialized to + proto and deserialized back via the dotted class_type path, not a + hardcoded per-store branch.""" + + def test_infra_object_survives_proto_round_trip(self): + original = DynamoDBTable( + name=f"{PROJECT}.view_a", region=REGION, endpoint_url=None + ) + infra = Infra(infra_objects=[original]) + + restored = Infra.from_proto(infra.to_proto()) + + assert len(restored.infra_objects) == 1 + restored_table = restored.infra_objects[0] + assert isinstance(restored_table, DynamoDBTable) + assert restored_table.name == f"{PROJECT}.view_a" + assert restored_table.region == REGION + + def test_round_trip_preserves_endpoint_url(self): + original = DynamoDBTable( + name=f"{PROJECT}.view_a", + region=REGION, + endpoint_url="http://localhost:8000", + ) + infra = Infra(infra_objects=[original]) + + restored = Infra.from_proto(infra.to_proto()) + + assert restored.infra_objects[0].endpoint_url == "http://localhost:8000"