feat: Populate created and updated timestamp on data sources

ntkathole · ntkathole · commit af3056b8a528 · 2025-07-12T15:43:35.000+05:30
Signed-off-by: ntkathole &lt;nikhilkathole2683@gmail.com&gt;
diff --git a/protos/feast/core/DataSource.proto b/protos/feast/core/DataSource.proto
@@ -95,6 +95,8 @@ message DataSource {
   message SourceMeta {
     google.protobuf.Timestamp earliestEventTimestamp = 1;
     google.protobuf.Timestamp latestEventTimestamp = 2;
+    google.protobuf.Timestamp created_timestamp = 3;
+    google.protobuf.Timestamp last_updated_timestamp = 4;
   }
 
   // Defines options for DataSource that sources features from a file
diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py
@@ -14,7 +14,7 @@
 import enum
 import warnings
 from abc import ABC, abstractmethod
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple
 
 from google.protobuf.duration_pb2 import Duration
@@ -27,6 +27,7 @@
 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
 from feast.repo_config import RepoConfig, get_data_source_class_from_type
 from feast.types import from_value_type
+from feast.utils import _utc_now
 from feast.value_type import ValueType
 
 
@@ -182,6 +183,8 @@ class DataSource(ABC):
         owner (optional): The owner of the data source, typically the email of the primary
             maintainer.
         date_partition_column (optional): Timestamp column used for partitioning. Not supported by all offline stores.
+        created_timestamp: The time when the data source was created.
+        last_updated_timestamp: The time when the data source was last updated.
     """
 
     name: str
@@ -192,6 +195,8 @@ class DataSource(ABC):
     tags: Dict[str, str]
     owner: str
     date_partition_column: str
+    created_timestamp: Optional[datetime]
+    last_updated_timestamp: Optional[datetime]
 
     def __init__(
         self,
@@ -242,6 +247,9 @@ def __init__(
         self.date_partition_column = (
             date_partition_column if date_partition_column else ""
         )
+        now = _utc_now()
+        self.created_timestamp = now
+        self.last_updated_timestamp = now
 
     def __hash__(self):
         return hash((self.name, self.timestamp_field))
@@ -295,15 +303,32 @@ def from_proto(data_source: DataSourceProto) -> Any:
 
         if data_source_type == DataSourceProto.SourceType.CUSTOM_SOURCE:
             cls = get_data_source_class_from_type(data_source.data_source_class_type)
-            return cls.from_proto(data_source)
-        cls = get_data_source_class_from_type(_DATA_SOURCE_OPTIONS[data_source_type])
-        return cls.from_proto(data_source)
+            data_source_instance = cls.from_proto(data_source)
+        else:
+            cls = get_data_source_class_from_type(
+                _DATA_SOURCE_OPTIONS[data_source_type]
+            )
+            data_source_instance = cls.from_proto(data_source)
+
+        data_source_instance._extract_timestamps_from_proto(data_source)
+
+        return data_source_instance
 
-    @abstractmethod
     def to_proto(self) -> DataSourceProto:
         """
         Converts a DataSourceProto object to its protobuf representation.
         """
+        proto = self._to_proto_impl()
+        self._set_timestamps_in_proto(proto)
+
+        return proto
+
+    @abstractmethod
+    def _to_proto_impl(self) -> DataSourceProto:
+        """
+        Subclass implementation of protobuf conversion.
+        This should be implemented by each DataSource subclass.
+        """
         raise NotImplementedError
 
     def validate(self, config: RepoConfig):
@@ -340,6 +365,42 @@ def get_table_query_string(self) -> str:
         """
         raise NotImplementedError
 
+    def _extract_timestamps_from_proto(self, data_source_proto: DataSourceProto):
+        """
+        Internal method to extract created_timestamp and last_updated_timestamp from protobuf.
+        Called automatically by the base from_proto method.
+        """
+        if data_source_proto.HasField("meta"):
+            if data_source_proto.meta.HasField("created_timestamp"):
+                self.created_timestamp = (
+                    data_source_proto.meta.created_timestamp.ToDatetime().replace(
+                        tzinfo=timezone.utc
+                    )
+                )
+            if data_source_proto.meta.HasField("last_updated_timestamp"):
+                self.last_updated_timestamp = (
+                    data_source_proto.meta.last_updated_timestamp.ToDatetime().replace(
+                        tzinfo=timezone.utc
+                    )
+                )
+
+    def _set_timestamps_in_proto(self, data_source_proto: DataSourceProto):
+        """
+        Internal method to set created_timestamp and last_updated_timestamp in protobuf.
+        Called automatically by the base to_proto method.
+        """
+        if not data_source_proto.HasField("meta"):
+            data_source_proto.meta.CopyFrom(DataSourceProto.SourceMeta())
+
+        if self.created_timestamp:
+            data_source_proto.meta.created_timestamp.FromDatetime(
+                self.created_timestamp
+            )
+        if self.last_updated_timestamp:
+            data_source_proto.meta.last_updated_timestamp.FromDatetime(
+                self.last_updated_timestamp
+            )
+
 
 @typechecked
 class KafkaSource(DataSource):
@@ -470,7 +531,7 @@ def from_proto(data_source: DataSourceProto):
             ),
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.STREAM_KAFKA,
@@ -485,6 +546,7 @@ def to_proto(self) -> DataSourceProto:
         data_source_proto.created_timestamp_column = self.created_timestamp_column
         if self.batch_source:
             data_source_proto.batch_source.MergeFrom(self.batch_source.to_proto())
+
         return data_source_proto
 
     def validate(self, config: RepoConfig):
@@ -587,7 +649,7 @@ def from_proto(data_source: DataSourceProto):
             owner=data_source.owner,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         schema_pb = []
 
         if isinstance(self.schema, Dict):
@@ -731,7 +793,7 @@ def __eq__(self, other):
     def __hash__(self):
         return super().__hash__()
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.STREAM_KINESIS,
@@ -829,7 +891,7 @@ def from_proto(data_source: DataSourceProto):
             owner=data_source.owner,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.PUSH_SOURCE,
diff --git a/sdk/python/feast/feature_view.py b/sdk/python/feast/feature_view.py
@@ -289,6 +289,7 @@ def __eq__(self, other):
             or self.batch_source != other.batch_source
             or self.stream_source != other.stream_source
             or sorted(self.entity_columns) != sorted(other.entity_columns)
+            or self.materialization_intervals != other.materialization_intervals
         ):
             return False
 
diff --git a/sdk/python/feast/infra/offline_stores/bigquery_source.py b/sdk/python/feast/infra/offline_stores/bigquery_source.py
@@ -120,7 +120,7 @@ def from_proto(data_source: DataSourceProto):
             owner=data_source.owner,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.BATCH_BIGQUERY,
diff --git a/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena_source.py b/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena_source.py
@@ -144,7 +144,7 @@ def data_source(self):
         """Returns the Athena data_source of this Athena source."""
         return self.athena_options.data_source
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         """
         Converts a RedshiftSource object to its protobuf representation.
 
diff --git a/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse_source.py b/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse_source.py
@@ -105,7 +105,7 @@ def from_proto(data_source: DataSourceProto) -> Any:
             owner=data_source.owner,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.CUSTOM_SOURCE,
diff --git a/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase_source.py b/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase_source.py
@@ -121,7 +121,7 @@ def from_proto(data_source: DataSourceProto):
             owner=data_source.owner,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.CUSTOM_SOURCE,
diff --git a/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssqlserver_source.py b/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssqlserver_source.py
@@ -220,7 +220,7 @@ def from_proto(data_source: DataSourceProto):
             date_partition_column=data_source.date_partition_column,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             type=DataSourceProto.CUSTOM_SOURCE,
             data_source_class_type="feast.infra.offline_stores.contrib.mssql_offline_store.mssqlserver_source.MsSqlServerSource",
diff --git a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres_source.py b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres_source.py
@@ -103,7 +103,7 @@ def from_proto(data_source: DataSourceProto):
             owner=data_source.owner,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.CUSTOM_SOURCE,
diff --git a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py
@@ -157,7 +157,7 @@ def from_proto(data_source: DataSourceProto) -> Any:
             owner=data_source.owner,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.BATCH_SPARK,
diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino_source.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino_source.py
@@ -198,7 +198,7 @@ def from_proto(data_source: DataSourceProto):
             owner=data_source.owner,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.BATCH_TRINO,
diff --git a/sdk/python/feast/infra/offline_stores/file_source.py b/sdk/python/feast/infra/offline_stores/file_source.py
@@ -131,7 +131,7 @@ def from_proto(data_source: DataSourceProto):
             owner=data_source.owner,
         )
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             name=self.name,
             type=DataSourceProto.BATCH_FILE,
diff --git a/sdk/python/feast/infra/offline_stores/redshift_source.py b/sdk/python/feast/infra/offline_stores/redshift_source.py
@@ -152,7 +152,7 @@ def database(self):
         """Returns the Redshift database of this Redshift source."""
         return self.redshift_options.database
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         """
         Converts a RedshiftSource object to its protobuf representation.
 
diff --git a/sdk/python/feast/infra/offline_stores/snowflake_source.py b/sdk/python/feast/infra/offline_stores/snowflake_source.py
@@ -164,7 +164,7 @@ def query(self):
         """Returns the snowflake options of this snowflake source."""
         return self.snowflake_options.query
 
-    def to_proto(self) -> DataSourceProto:
+    def _to_proto_impl(self) -> DataSourceProto:
         """
         Converts a SnowflakeSource object to its protobuf representation.
 
diff --git a/sdk/python/feast/infra/registry/base_registry.py b/sdk/python/feast/infra/registry/base_registry.py
@@ -31,6 +31,7 @@
 from feast.permissions.permission import Permission
 from feast.project import Project
 from feast.project_metadata import ProjectMetadata
+from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
 from feast.protos.feast.core.Entity_pb2 import Entity as EntityProto
 from feast.protos.feast.core.FeatureService_pb2 import (
     FeatureService as FeatureServiceProto,
@@ -1057,4 +1058,6 @@ def deserialize_registry_values(serialized_proto, feast_obj_type) -> Any:
             return PermissionProto.FromString(serialized_proto)
         if feast_obj_type == Project:
             return ProjectProto.FromString(serialized_proto)
+        if issubclass(feast_obj_type, DataSource):
+            return DataSourceProto.FromString(serialized_proto)
         return None
diff --git a/sdk/python/feast/infra/registry/registry.py b/sdk/python/feast/infra/registry/registry.py
@@ -338,10 +338,27 @@ def list_data_sources(
     def apply_data_source(
         self, data_source: DataSource, project: str, commit: bool = True
     ):
+        now = _utc_now()
+        if not data_source.created_timestamp:
+            data_source.created_timestamp = now
+        data_source.last_updated_timestamp = now
+
         registry = self._prepare_registry_for_changes(project)
+
         for idx, existing_data_source_proto in enumerate(registry.data_sources):
             if existing_data_source_proto.name == data_source.name:
-                del registry.data_sources[idx]
+                existing_data_source = DataSource.from_proto(existing_data_source_proto)
+                # Check if the data source has actually changed
+                if existing_data_source == data_source:
+                    return
+                else:
+                    # Preserve created_timestamp from existing data source
+                    data_source.created_timestamp = (
+                        existing_data_source.created_timestamp
+                    )
+                    del registry.data_sources[idx]
+                    break
+
         data_source_proto = data_source.to_proto()
         data_source_proto.project = project
         data_source_proto.data_source_class_type = (
diff --git a/sdk/python/feast/infra/registry/sql.py b/sdk/python/feast/infra/registry/sql.py
@@ -939,8 +939,6 @@ def _apply_object(
                 getattr(table.c, id_field_name) == name, table.c.project_id == project
             )
             row = conn.execute(stmt).first()
-            if hasattr(obj, "last_updated_timestamp"):
-                obj.last_updated_timestamp = update_datetime
 
             if row:
                 if proto_field_name in [
@@ -950,21 +948,30 @@ def _apply_object(
                     "feature_service_proto",
                     "permission_proto",
                     "project_proto",
+                    "data_source_proto",
                 ]:
                     deserialized_proto = self.deserialize_registry_values(
                         row._mapping[proto_field_name], type(obj)
                     )
-                    obj.created_timestamp = (
-                        deserialized_proto.meta.created_timestamp.ToDatetime().replace(
+                    if deserialized_proto is not None:
+                        # Check if the object has actually changed (same as feature views)
+                        existing_obj = type(obj).from_proto(deserialized_proto)
+                        if existing_obj == obj:
+                            return  # No changes, exit early
+
+                        # Object has changed, preserve created_timestamp, update last_updated_timestamp
+                        obj.created_timestamp = deserialized_proto.meta.created_timestamp.ToDatetime().replace(
                             tzinfo=timezone.utc
                         )
-                    )
-                    if isinstance(obj, (FeatureView, StreamFeatureView)):
-                        obj.update_materialization_intervals(
-                            type(obj)
-                            .from_proto(deserialized_proto)
-                            .materialization_intervals
-                        )
+                        if hasattr(obj, "last_updated_timestamp"):
+                            obj.last_updated_timestamp = update_datetime
+                        if isinstance(obj, (FeatureView, StreamFeatureView)):
+                            obj.update_materialization_intervals(
+                                type(obj)
+                                .from_proto(deserialized_proto)
+                                .materialization_intervals
+                            )
+
                 values = {
                     proto_field_name: obj.to_proto().SerializeToString(),
                     "last_updated_timestamp": update_time,
@@ -981,6 +988,12 @@ def _apply_object(
                 )
                 conn.execute(update_stmt)
             else:
+                # Creating new object - set timestamps consistently for all objects
+                if hasattr(obj, "created_timestamp"):
+                    obj.created_timestamp = update_datetime
+                if hasattr(obj, "last_updated_timestamp"):
+                    obj.last_updated_timestamp = update_datetime
+
                 obj_proto = obj.to_proto()
 
                 if hasattr(obj_proto, "meta") and hasattr(
diff --git a/sdk/python/feast/protos/feast/core/DataSource_pb2.py b/sdk/python/feast/protos/feast/core/DataSource_pb2.py
diff --git a/sdk/python/feast/protos/feast/core/DataSource_pb2.pyi b/sdk/python/feast/protos/feast/core/DataSource_pb2.pyi
diff --git a/sdk/python/tests/integration/registration/test_universal_registry.py b/sdk/python/tests/integration/registration/test_universal_registry.py
diff --git a/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py b/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py
diff --git a/sdk/python/tests/unit/test_data_sources.py b/sdk/python/tests/unit/test_data_sources.py

Original file line number	Diff line number	Diff line change
`@@ -95,6 +95,8 @@ message DataSource {`
`95`	`95`	`message SourceMeta {`
`96`	`96`	`google.protobuf.Timestamp earliestEventTimestamp = 1;`
`97`	`97`	`google.protobuf.Timestamp latestEventTimestamp = 2;`
	`98`	`+ google.protobuf.Timestamp created_timestamp = 3;`
	`99`	`+ google.protobuf.Timestamp last_updated_timestamp = 4;`
`98`	`100`	`}`
`99`	`101`
`100`	`102`	`// Defines options for DataSource that sources features from a file`
Original file line number	Diff line number	Diff line change
`@@ -120,7 +120,7 @@ def from_proto(data_source: DataSourceProto):`
`120`	`120`	`owner=data_source.owner,`
`121`	`121`	`)`
`122`	`122`
`123`		`- def to_proto(self) -> DataSourceProto:`
	`123`	`+ def _to_proto_impl(self) -> DataSourceProto:`
`124`	`124`	`data_source_proto = DataSourceProto(`
`125`	`125`	`name=self.name,`
`126`	`126`	`type=DataSourceProto.BATCH_BIGQUERY,`
Original file line number	Diff line number	Diff line change
`@@ -105,7 +105,7 @@ def from_proto(data_source: DataSourceProto) -> Any:`
`105`	`105`	`owner=data_source.owner,`
`106`	`106`	`)`
`107`	`107`
`108`		`- def to_proto(self) -> DataSourceProto:`
	`108`	`+ def _to_proto_impl(self) -> DataSourceProto:`
`109`	`109`	`data_source_proto = DataSourceProto(`
`110`	`110`	`name=self.name,`
`111`	`111`	`type=DataSourceProto.CUSTOM_SOURCE,`
Original file line number	Diff line number	Diff line change
`@@ -121,7 +121,7 @@ def from_proto(data_source: DataSourceProto):`
`121`	`121`	`owner=data_source.owner,`
`122`	`122`	`)`
`123`	`123`
`124`		`- def to_proto(self) -> DataSourceProto:`
	`124`	`+ def _to_proto_impl(self) -> DataSourceProto:`
`125`	`125`	`data_source_proto = DataSourceProto(`
`126`	`126`	`name=self.name,`
`127`	`127`	`type=DataSourceProto.CUSTOM_SOURCE,`
Original file line number	Diff line number	Diff line change
`@@ -220,7 +220,7 @@ def from_proto(data_source: DataSourceProto):`
`220`	`220`	`date_partition_column=data_source.date_partition_column,`
`221`	`221`	`)`
`222`	`222`
`223`		`- def to_proto(self) -> DataSourceProto:`
	`223`	`+ def _to_proto_impl(self) -> DataSourceProto:`
`224`	`224`	`data_source_proto = DataSourceProto(`
`225`	`225`	`type=DataSourceProto.CUSTOM_SOURCE,`
`226`	`226`	`data_source_class_type="feast.infra.offline_stores.contrib.mssql_offline_store.mssqlserver_source.MsSqlServerSource",`
Original file line number	Diff line number	Diff line change
`@@ -103,7 +103,7 @@ def from_proto(data_source: DataSourceProto):`
`103`	`103`	`owner=data_source.owner,`
`104`	`104`	`)`
`105`	`105`
`106`		`- def to_proto(self) -> DataSourceProto:`
	`106`	`+ def _to_proto_impl(self) -> DataSourceProto:`
`107`	`107`	`data_source_proto = DataSourceProto(`
`108`	`108`	`name=self.name,`
`109`	`109`	`type=DataSourceProto.CUSTOM_SOURCE,`
Original file line number	Diff line number	Diff line change
`@@ -157,7 +157,7 @@ def from_proto(data_source: DataSourceProto) -> Any:`
`157`	`157`	`owner=data_source.owner,`
`158`	`158`	`)`
`159`	`159`
`160`		`- def to_proto(self) -> DataSourceProto:`
	`160`	`+ def _to_proto_impl(self) -> DataSourceProto:`
`161`	`161`	`data_source_proto = DataSourceProto(`
`162`	`162`	`name=self.name,`
`163`	`163`	`type=DataSourceProto.BATCH_SPARK,`