feat: Event timestamps response

sdk/python/feast/feature_store.py

@@ -1508,10 +1508,10 @@ def _read_from_online_store(
 
         # Each row is a set of features for a given entity key. We only need to convert
         # the data to Protobuf once.
-        row_ts_proto = Timestamp()
         null_value = Value()
         read_row_protos = []
         for read_row in read_rows:
+            row_ts_proto = Timestamp()
             row_ts, feature_data = read_row
             if row_ts is not None:
                 row_ts_proto.FromDatetime(row_ts)

sdk/python/feast/infra/online_stores/redis.py

@@ -27,6 +27,7 @@
     Union,
 )
 
+import pytz
 from google.protobuf.timestamp_pb2 import Timestamp
 from pydantic import StrictStr
 from pydantic.typing import Literal
@@ -302,5 +303,5 @@ def _get_features_for_entity(
         if not res:
             return None, None
         else:
-            timestamp = datetime.fromtimestamp(res_ts.seconds)
+            timestamp = datetime.fromtimestamp(res_ts.seconds, tz=pytz.utc)
             return timestamp, res

sdk/python/feast/online_response.py

@@ -23,7 +23,7 @@
 
 class OnlineResponse:
     """
-    Defines a online response in feast.
+    Defines an online response in feast.
     """
 
     def __init__(self, online_response_proto: GetOnlineFeaturesResponse):
@@ -66,3 +66,28 @@ def to_df(self) -> pd.DataFrame:
         """
 
         return pd.DataFrame(self.to_dict())
+
+    def event_timestamps_dict(self) -> Dict[str, List[int]]:
+        """
+        Converts GetOnlineFeaturesResponse feature event timestamps into a dictionary form.
+        Includes the entity id
+        """
+        response: Dict[str, List[int]] = {}
+
+        for result in self.proto.results:
+            for idx, feature_ref in enumerate(self.proto.metadata.feature_names.val):
+                # obtain the entity id
+                if idx == 0:
+                    value = feast_value_type_to_python_type(result.values[idx])
+                # feature timestamps
+                else:
+                    value = result.event_timestamps[idx].seconds
+
+                if feature_ref not in response:
+                    response[feature_ref] = [value]
+                else:
+                    response[feature_ref].append(value)
+        return response
+
+    def event_timestamps_df(self) -> pd.DataFrame:
+        return pd.DataFrame(self.event_timestamps_dict())

sdk/python/tests/integration/online_store/test_universal_online.py

@@ -322,6 +322,58 @@ def get_online_features_dict(
     return dict1
 
 
+@pytest.mark.integration
+@pytest.mark.universal
+@pytest.mark.parametrize("full_feature_names", [True, False], ids=lambda v: str(v))
+def test_online_retrieval_timestamps(
+    environment, universal_data_sources, full_feature_names
+):
+    fs = environment.feature_store
+    entities, datasets, data_sources = universal_data_sources
+    feature_views = construct_universal_feature_views(data_sources)
+
+    fs.apply([driver(), feature_views["driver"], feature_views["global"]])
+
+    # fake data to ingest into Online Store
+    data = {
+        "driver_id": [1, 2],
+        "conv_rate": [0.5, 0.3],
+        "acc_rate": [0.6, 0.4],
+        "avg_daily_trips": [4, 5],
+        "event_timestamp": [
+            pd.to_datetime(1646263500, utc=True, unit="s"),
+            pd.to_datetime(1646263600, utc=True, unit="s"),
+        ],
+        "created": [
+            pd.to_datetime(1646263500, unit="s"),
+            pd.to_datetime(1646263600, unit="s"),
+        ],
+    }
+    df_ingest = pd.DataFrame(data)
+
+    # directly ingest data into the Online Store
+    fs.write_to_online_store("driver_stats", df_ingest)
+
+    response = fs.get_online_features(
+        features=[
+            "driver_stats:avg_daily_trips",
+            "driver_stats:acc_rate",
+            "driver_stats:conv_rate",
+        ],
+        entity_rows=[{"driver": 1}, {"driver": 2}],
+    )
+    df = response.event_timestamps_df()
+    assertpy.assert_that(len(df)).is_equal_to(2)
+    assertpy.assert_that(df["driver_id"].iloc[0]).is_equal_to(1)
+    assertpy.assert_that(df["driver_id"].iloc[1]).is_equal_to(2)
+    assertpy.assert_that(df["avg_daily_trips"].iloc[0]).is_equal_to(1646263500)
+    assertpy.assert_that(df["avg_daily_trips"].iloc[1]).is_equal_to(1646263600)
+    assertpy.assert_that(df["acc_rate"].iloc[0]).is_equal_to(1646263500)
+    assertpy.assert_that(df["acc_rate"].iloc[1]).is_equal_to(1646263600)
+    assertpy.assert_that(df["conv_rate"].iloc[0]).is_equal_to(1646263500)
+    assertpy.assert_that(df["conv_rate"].iloc[1]).is_equal_to(1646263600)
+
+
 @pytest.mark.integration
 @pytest.mark.universal
 @pytest.mark.parametrize("full_feature_names", [True, False], ids=lambda v: str(v))