feat: Feast/IKV datetime edgecase errors (#4211)

* feat: Feast/IKV datetime edgecase errors

Signed-off-by: Pushkar Gupta <pushkar.moi@gmail.com>

* linter

Signed-off-by: Pushkar Gupta <pushkar.moi@gmail.com>

* linter

Signed-off-by: Pushkar Gupta <pushkar.moi@gmail.com>

---------

Signed-off-by: Pushkar Gupta <pushkar.moi@gmail.com>

sdk/python/feast/infra/online_stores/contrib/ikv_online_store/ikv.py

@@ -11,6 +11,8 @@
     Tuple,
 )
 
+import pytz
+from google.protobuf.timestamp_pb2 import Timestamp
 from ikvpy.client import IKVReader, IKVWriter
 from ikvpy.clientoptions import ClientOptions, ClientOptionsBuilder
 from ikvpy.document import IKVDocument, IKVDocumentBuilder
@@ -162,7 +164,9 @@ def _decode_fields_for_primary_key(
         dt: Optional[datetime] = None
         dt_bytes = next(value_iter)
         if dt_bytes:
-            dt = datetime.fromisoformat(str(dt_bytes, "utf-8"))
+            proto_timestamp = Timestamp()
+            proto_timestamp.ParseFromString(dt_bytes)
+            dt = datetime.fromtimestamp(proto_timestamp.seconds, tz=pytz.utc)
 
         # decode other features
         features = {}
@@ -252,12 +256,17 @@ def _create_document(
         """Converts feast key-value pairs into an IKV document."""
 
         # initialie builder by inserting primary key and row creation timestamp
-        event_timestamp_str: str = utils.make_tzaware(event_timestamp).isoformat()
+        event_timestamp_seconds = int(utils.make_tzaware(event_timestamp).timestamp())
+        event_timestamp_seconds_proto = Timestamp()
+        event_timestamp_seconds_proto.seconds = event_timestamp_seconds
+
+        # event_timestamp_str: str = utils.make_tzaware(event_timestamp).isoformat()
         builder = (
             IKVDocumentBuilder()
             .put_string_field(PRIMARY_KEY_FIELD_NAME, entity_id)
             .put_bytes_field(
-                EVENT_CREATION_TIMESTAMP_FIELD_NAME, event_timestamp_str.encode("utf-8")
+                EVENT_CREATION_TIMESTAMP_FIELD_NAME,
+                event_timestamp_seconds_proto.SerializeToString(),
             )
         )