feat: Durable materialization-interval history for every registry backend

protos/feast/core/FeatureView.proto

@@ -109,6 +109,23 @@ message MaterializationInterval {
     google.protobuf.Timestamp end_time = 2;
 }
 
+// A single durably-retained entry in a feature view's full, uncapped
+// materialization-interval history. Unlike MaterializationInterval (which
+// lives on a specific FeatureViewMeta and is capped to the most recent N
+// entries), this is discriminated by feature_view_name/project so it can
+// live in a separate collection -- a registry-backend-specific table (SQL,
+// Snowflake) or a top-level Registry field (file-based) -- and is written
+// for every interval, including ones already dropped from the capped list.
+message MaterializationIntervalHistoryEntry {
+    string feature_view_name = 1;
+    string project = 2;
+    google.protobuf.Timestamp start_time = 3;
+    google.protobuf.Timestamp end_time = 4;
+    // When this entry was recorded to history (not necessarily when the
+    // materialization run itself happened).
+    google.protobuf.Timestamp recorded_at = 5;
+}
+
 message FeatureViewList {
     repeated FeatureView featureviews = 1;
 }

protos/feast/core/Registry.proto

@@ -36,7 +36,7 @@ import "google/protobuf/timestamp.proto";
 import "feast/core/Permission.proto";
 import "feast/core/Project.proto";
 
-// Next id: 18
+// Next id: 19
 message Registry {
     repeated Entity entities = 1;
     repeated FeatureTable feature_tables = 2;
@@ -57,6 +57,10 @@ message Registry {
     google.protobuf.Timestamp last_updated = 5;
     repeated Permission permissions = 16;
     repeated Project projects = 17;
+    // Full, uncapped history of materialization intervals across every feature
+    // view in every project -- the file-based registry's counterpart to the
+    // SQL/Snowflake backends' separate materialization_interval_history table.
+    repeated MaterializationIntervalHistoryEntry materialization_interval_history = 18;
 }
 
 message ProjectMetadata {

protos/feast/registry/RegistryServer.proto

@@ -87,6 +87,7 @@ service RegistryServer{
   rpc DeleteProject (DeleteProjectRequest) returns (google.protobuf.Empty) {}
 
   rpc ApplyMaterialization (ApplyMaterializationRequest) returns (google.protobuf.Empty) {}
+  rpc GetMaterializationIntervalHistory (GetMaterializationIntervalHistoryRequest) returns (GetMaterializationIntervalHistoryResponse) {}
   rpc ListProjectMetadata (ListProjectMetadataRequest) returns (ListProjectMetadataResponse) {}
   rpc UpdateInfra (UpdateInfraRequest) returns (google.protobuf.Empty) {}
   rpc GetInfra (GetInfraRequest) returns (feast.core.Infra) {}
@@ -159,6 +160,18 @@ message ApplyMaterializationRequest {
   bool commit = 5;
 }
 
+message GetMaterializationIntervalHistoryRequest {
+    string feature_view_name = 1;
+    string project = 2;
+    PaginationParams pagination = 3;
+    SortingParams sorting = 4;
+}
+
+message GetMaterializationIntervalHistoryResponse {
+    repeated feast.core.MaterializationIntervalHistoryEntry entries = 1;
+    PaginationMetadata pagination = 2;
+}
+
 message ApplyEntityRequest {
   feast.core.Entity entity = 1;
   string project = 2;

sdk/python/feast/expediagroup/pydantic_models/materialization_interval_history_model.py

@@ -0,0 +1,62 @@
+"""
+Pydantic Model for MaterializationIntervalHistoryEntry
+
+Copyright 2023 Expedia Group
+"""
+
+from datetime import datetime, timezone
+
+from pydantic import BaseModel, ConfigDict
+from typing_extensions import Self
+
+from feast.protos.feast.core.FeatureView_pb2 import (
+    MaterializationIntervalHistoryEntry as MaterializationIntervalHistoryEntryProto,
+)
+
+
+class MaterializationIntervalHistoryEntryModel(BaseModel):
+    """
+    Pydantic model of a single entry in a feature view's full, uncapped
+    materialization-interval history. Unlike most sibling models, there is
+    no intermediate domain object for this type (it's a plain historical
+    record, not a first-class Feast object) -- this converts directly
+    to/from the proto message.
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")
+
+    feature_view_name: str
+    project: str
+    start_time: datetime
+    end_time: datetime
+    recorded_at: datetime
+
+    def to_proto(self) -> MaterializationIntervalHistoryEntryProto:
+        """
+        Converts this model to its protobuf representation.
+        """
+        entry = MaterializationIntervalHistoryEntryProto(
+            feature_view_name=self.feature_view_name,
+            project=self.project,
+        )
+        entry.start_time.FromDatetime(self.start_time)
+        entry.end_time.FromDatetime(self.end_time)
+        entry.recorded_at.FromDatetime(self.recorded_at)
+        return entry
+
+    @classmethod
+    def from_proto(
+        cls,
+        proto: MaterializationIntervalHistoryEntryProto,
+    ) -> Self:  # type: ignore
+        """
+        Converts a MaterializationIntervalHistoryEntry proto to its pydantic
+        model representation.
+        """
+        return cls(
+            feature_view_name=proto.feature_view_name,
+            project=proto.project,
+            start_time=proto.start_time.ToDatetime().replace(tzinfo=timezone.utc),
+            end_time=proto.end_time.ToDatetime().replace(tzinfo=timezone.utc),
+            recorded_at=proto.recorded_at.ToDatetime().replace(tzinfo=timezone.utc),
+        )

sdk/python/feast/feature_view.py

@@ -62,6 +62,20 @@
 
 ONLINE_STORE_TAG_SUFFIX = "online_store_"
 
+# Rolling-window cap on FeatureViewMeta.materialization_intervals -- kept
+# small so the registry's per-feature-view proto blob doesn't grow
+# unboundedly over the life of a feature view; this field is only ever meant
+# to answer "when was this feature view most recently materialized," which
+# never needs more than a handful of recent entries. The full, uncapped
+# history (including every interval this field ever drops) is durably
+# retained by each registry backend in a separate materialization-interval
+# history store -- see add_materialization_interval/_cap_materialization_intervals's
+# return value, which callers archive there. Configurable per registry
+# instance via RegistryConfig.materialization_intervals_max_len (currently
+# honored by SqlRegistry/SqlFallbackRegistry only) to stage a rollout of a
+# new cap value via config rather than a code change.
+MATERIALIZATION_INTERVALS_MAX_LEN = 10
+
 logger = logging.getLogger(__name__)
 
 
@@ -389,14 +403,83 @@ def with_join_key_map(self, join_key_map: Dict[str, str]):
         return cp
 
     def update_materialization_intervals(
-        self, existing_materialization_intervals: List[Tuple[datetime, datetime]]
-    ):
+        self,
+        existing_materialization_intervals: List[Tuple[datetime, datetime]],
+        max_intervals: Optional[int] = None,
+    ) -> List[Tuple[datetime, datetime]]:
         if (
             len(existing_materialization_intervals) > 0
             and len(self.materialization_intervals) == 0
         ):
             for interval in existing_materialization_intervals:
                 self.materialization_intervals.append((interval[0], interval[1]))
+        # Defensively re-apply the cap here too: this hydrates an in-memory
+        # object from what's already persisted (e.g. the SQL registry's
+        # "no changes, preserve existing intervals" path), so an
+        # already-persisted, not-yet-migrated feature view with more than
+        # the cap can't leak an over-long list back out through this path.
+        # The caller is responsible for archiving whatever this returns --
+        # this is also the moment a feature view that already has more than
+        # the cap (e.g. persisted before a cap tightening rollout) first gets
+        # trimmed, so the returned list may be more than one entry.
+        return self._cap_materialization_intervals(max_intervals)
+
+    def add_materialization_interval(
+        self,
+        start_date: datetime,
+        end_date: datetime,
+        max_intervals: Optional[int] = None,
+    ) -> List[Tuple[datetime, datetime]]:
+        """
+        Records that this feature view was materialized for [start_date, end_date),
+        keeping only the most recent `max_intervals` entries (or
+        MATERIALIZATION_INTERVALS_MAX_LEN if not given).
+
+        This is the single choke point every registry backend's
+        apply_materialization should call through (instead of appending to
+        materialization_intervals directly), so the cap is enforced
+        identically everywhere.
+
+        Returns the list of intervals dropped by the cap as a result of this
+        call (usually empty). Callers are expected to durably archive both
+        the dropped intervals and the newly-added (start_date, end_date) to a
+        separate, uncapped materialization-interval history store -- this
+        object only ever holds the bounded, recent-only view.
+
+        max_intervals lets a caller override the default cap -- e.g. a
+        registry backend reading RegistryConfig.materialization_intervals_max_len
+        so a rollout of a new cap value can be staged via config rather than
+        a code change (start high to match today's effectively-uncapped
+        behavior, then lower it in steps).
+        """
+        self.materialization_intervals.append((start_date, end_date))
+        return self._cap_materialization_intervals(max_intervals)
+
+    def _cap_materialization_intervals(
+        self, max_intervals: Optional[int] = None
+    ) -> List[Tuple[datetime, datetime]]:
+        effective_max = (
+            max_intervals
+            if max_intervals is not None
+            else MATERIALIZATION_INTERVALS_MAX_LEN
+        )
+        # Guard against effective_max <= 0: `-effective_max` would be `0` (or
+        # positive) rather than a negative slice bound, so `list[:-0]` and
+        # `list[-0:]` both silently resolve to slicing at index 0 -- meaning
+        # a cap of 0 would report nothing as dropped and leave the list
+        # completely untouched instead of capping it to empty. A cap of 0
+        # (or a nonsensical negative value) means "keep none".
+        if effective_max <= 0:
+            dropped = self.materialization_intervals
+            self.materialization_intervals = []
+            return dropped
+        if len(self.materialization_intervals) > effective_max:
+            dropped = self.materialization_intervals[:-effective_max]
+            self.materialization_intervals = self.materialization_intervals[
+                -effective_max:
+            ]
+            return dropped
+        return []
 
     def to_proto(self) -> FeatureViewProto:
         """

sdk/python/feast/infra/registry/base_registry.py

@@ -37,6 +37,9 @@
     FeatureService as FeatureServiceProto,
 )
 from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto
+from feast.protos.feast.core.FeatureView_pb2 import (
+    MaterializationIntervalHistoryEntry as MaterializationIntervalHistoryEntryProto,
+)
 from feast.protos.feast.core.OnDemandFeatureView_pb2 import (
     OnDemandFeatureView as OnDemandFeatureViewProto,
 )
@@ -494,6 +497,31 @@ def apply_materialization(
         """
         raise NotImplementedError
 
+    @abstractmethod
+    def get_materialization_interval_history(
+        self,
+        feature_view_name: str,
+        project: str,
+    ) -> List[MaterializationIntervalHistoryEntryProto]:
+        """
+        Retrieves the full, uncapped history of materialization intervals for a
+        feature view -- including entries already dropped from
+        FeatureViewMeta.materialization_intervals by the
+        materialization_intervals_max_len cap. Every registry backend durably
+        retains this history in a separate table/collection from the capped
+        list (see apply_materialization), so this works the same way
+        regardless of registry backend.
+
+        Args:
+            feature_view_name: Name of the feature view to fetch history for
+            project: Feast project that this feature view belongs to
+
+        Returns:
+            List of MaterializationIntervalHistoryEntry protos, ordered by
+            start_time ascending
+        """
+        raise NotImplementedError
+
     # Saved dataset operations
     @abstractmethod
     def apply_saved_dataset(

sdk/python/feast/infra/registry/http.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import threading
 import time
@@ -34,6 +35,9 @@
     OnDemandFeatureViewModel,
     SortedFeatureViewModel,
 )
+from feast.expediagroup.pydantic_models.materialization_interval_history_model import (
+    MaterializationIntervalHistoryEntryModel,
+)
 from feast.expediagroup.pydantic_models.project_metadata_model import (
     ProjectMetadataModel,
 )
@@ -666,28 +670,52 @@ def apply_materialization(
     ):
         if isinstance(feature_view, OnDemandFeatureView):
             raise TypeError("Materialization not supported for OnDemandFeatureView")
+        if not isinstance(feature_view, (FeatureView, SortedFeatureView)):
+            raise TypeError(
+                "Unsupported FeatureView type. Please use either FeatureView, SortedFeatureView or OnDemandFeatureView only"
+            )
         try:
-            feature_view.materialization_intervals.append((start_date, end_date))
+            # A dedicated endpoint, rather than the generic whole-object
+            # feature_views PUT: this lets the server fetch the canonical
+            # stored feature view and call apply_materialization directly,
+            # so the cap/archive logic in SqlFallbackRegistry.apply_materialization
+            # actually runs -- appending the interval client-side and PUTting
+            # the whole object (the old approach) bypassed that entirely,
+            # and could clobber previously-stored intervals if this
+            # in-memory object wasn't fully hydrated.
+            url = (
+                f"{self.base_url}/projects/{project}/feature_views/"
+                f"{feature_view.name}/materialization_intervals"
+            )
             params = {"commit": commit}
-            url = f"{self.base_url}/projects/{project}/feature_views"
-            if isinstance(feature_view, SortedFeatureView):
-                data = SortedFeatureViewModel.from_feature_view(
-                    feature_view
-                ).model_dump_json()
-                response_data = self._send_request("PUT", url, params=params, data=data)
-                return SortedFeatureViewModel.model_validate(
-                    response_data
-                ).to_feature_view()
-            elif isinstance(feature_view, FeatureView):
-                data = FeatureViewModel.from_feature_view(
-                    feature_view
-                ).model_dump_json()
-                response_data = self._send_request("PUT", url, params=params, data=data)
-                return FeatureViewModel.model_validate(response_data).to_feature_view()
-            else:
-                raise TypeError(
-                    "Unsupported FeatureView type. Please use either FeatureView, SortedFeatureView or OnDemandFeatureView only"
-                )
+            data = json.dumps(
+                {
+                    "start_date": start_date.isoformat(),
+                    "end_date": end_date.isoformat(),
+                }
+            )
+            self._send_request("PUT", url, params=params, data=data)
+        except Exception as exception:
+            self._handle_exception(exception)
+
+    def get_materialization_interval_history(
+        self,
+        feature_view_name: str,
+        project: str,
+    ):
+        try:
+            url = (
+                f"{self.base_url}/projects/{project}/feature_views/"
+                f"{feature_view_name}/materialization_interval_history"
+            )
+            response_data = self._send_request("GET", url)
+            response_list = response_data if isinstance(response_data, list) else []
+            return [
+                MaterializationIntervalHistoryEntryModel.model_validate(
+                    entry
+                ).to_proto()
+                for entry in response_list
+            ]
         except Exception as exception:
             self._handle_exception(exception)