We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Datastore online request makes a call once for each entity
**Is your feature request related to a problem? Please describe.**
There are performance issues with Datastore when fetching features for multiple entities. An individual fetch request is made for each entity 

**Describe the solution you'd like**
I would like to have a more efficient access to the Datastore for multiple entities. The python api for datastore offers a method `get_multi()` which allows to fetch multiple keys in one request.

**Describe alternatives you've considered**
---

**Additional context**

I have done some basic tests using the `get_multi` vs get and it seems that a improvement can be made by using the function `get_multi`

Here is part of code  I used for testing:

```
import numpy as np
import timeit
from google.cloud import datastore

def multi_online_read(num_keys=5):

    client = datastore.Client(
                    project=project_name, namespace=None,
                )
    
    feast_project = "feature_repo"
    key = client.key("Project", feast_project, "Table", table, "Row",  row)
    keys = []
    response = []
    for i in range(num_keys):
        keys.append(key)

    response = client.get_multi(keys)

    return response

def online_read(num_keys=5):
    client = datastore.Client(
                    project=project_name, namespace=None,
                )
    
    feast_project = "feature_repo"
    key = client.key("Project", feast_project, "Table", table, "Row", row)
    
    response = []
    for i in range(num_keys):
        response.append(client.get(key))


    return response

if __name__ == "__main__":

    results = timeit.repeat(online_read, repeat=10, number=1)
    print(results)
    print(np.mean(results))

    results = timeit.repeat(multi_online_read, repeat=10, number=1)
    print(results)
    print(np.mean(results))
```






--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/feast/infra/online_stores/datastore.py`
Content:
```
1 # Copyright 2021 The Feast Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     https://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 import itertools
15 from datetime import datetime
16 from multiprocessing.pool import ThreadPool
17 from typing import Any, Callable, Dict, Iterator, List, Optional, Sequence, Tuple, Union
18 
19 from pydantic import PositiveInt, StrictStr
20 from pydantic.typing import Literal
21 
22 from feast import Entity, FeatureTable, utils
23 from feast.feature_view import FeatureView
24 from feast.infra.online_stores.helpers import compute_entity_id
25 from feast.infra.online_stores.online_store import OnlineStore
26 from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
27 from feast.protos.feast.types.Value_pb2 import Value as ValueProto
28 from feast.repo_config import FeastConfigBaseModel, RepoConfig
29 
30 try:
31     from google.auth.exceptions import DefaultCredentialsError
32     from google.cloud import datastore
33 except ImportError as e:
34     from feast.errors import FeastExtrasDependencyImportError, FeastProviderLoginError
35 
36     raise FeastExtrasDependencyImportError("gcp", str(e))
37 
38 
39 ProtoBatch = Sequence[
40     Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]
41 ]
42 
43 
44 class DatastoreOnlineStoreConfig(FeastConfigBaseModel):
45     """ Online store config for GCP Datastore """
46 
47     type: Literal["datastore"] = "datastore"
48     """ Online store type selector"""
49 
50     project_id: Optional[StrictStr] = None
51     """ (optional) GCP Project Id """
52 
53     namespace: Optional[StrictStr] = None
54     """ (optional) Datastore namespace """
55 
56     write_concurrency: Optional[PositiveInt] = 40
57     """ (optional) Amount of threads to use when writing batches of feature rows into Datastore"""
58 
59     write_batch_size: Optional[PositiveInt] = 50
60     """ (optional) Amount of feature rows per batch being written into Datastore"""
61 
62 
63 class DatastoreOnlineStore(OnlineStore):
64     """
65     OnlineStore is an object used for all interaction between Feast and the service used for offline storage of
66     features.
67     """
68 
69     _client: Optional[datastore.Client] = None
70 
71     def update(
72         self,
73         config: RepoConfig,
74         tables_to_delete: Sequence[Union[FeatureTable, FeatureView]],
75         tables_to_keep: Sequence[Union[FeatureTable, FeatureView]],
76         entities_to_delete: Sequence[Entity],
77         entities_to_keep: Sequence[Entity],
78         partial: bool,
79     ):
80         """
81         """
82         online_config = config.online_store
83         assert isinstance(online_config, DatastoreOnlineStoreConfig)
84         client = self._get_client(online_config)
85         feast_project = config.project
86 
87         for table in tables_to_keep:
88             key = client.key("Project", feast_project, "Table", table.name)
89             entity = datastore.Entity(
90                 key=key, exclude_from_indexes=("created_ts", "event_ts", "values")
91             )
92             entity.update({"created_ts": datetime.utcnow()})
93             client.put(entity)
94 
95         for table in tables_to_delete:
96             _delete_all_values(
97                 client, client.key("Project", feast_project, "Table", table.name)
98             )
99 
100             # Delete the table metadata datastore entity
101             key = client.key("Project", feast_project, "Table", table.name)
102             client.delete(key)
103 
104     def teardown(
105         self,
106         config: RepoConfig,
107         tables: Sequence[Union[FeatureTable, FeatureView]],
108         entities: Sequence[Entity],
109     ):
110         """
111         There's currently no teardown done for Datastore.
112         """
113         online_config = config.online_store
114         assert isinstance(online_config, DatastoreOnlineStoreConfig)
115         client = self._get_client(online_config)
116         feast_project = config.project
117 
118         for table in tables:
119             _delete_all_values(
120                 client, client.key("Project", feast_project, "Table", table.name)
121             )
122 
123             # Delete the table metadata datastore entity
124             key = client.key("Project", feast_project, "Table", table.name)
125             client.delete(key)
126 
127     def _get_client(self, online_config: DatastoreOnlineStoreConfig):
128 
129         if not self._client:
130             try:
131                 self._client = datastore.Client(
132                     project=online_config.project_id, namespace=online_config.namespace,
133                 )
134             except DefaultCredentialsError as e:
135                 raise FeastProviderLoginError(
136                     str(e)
137                     + '\nIt may be necessary to run "gcloud auth application-default login" if you would like to use your '
138                     "local Google Cloud account "
139                 )
140         return self._client
141 
142     def online_write_batch(
143         self,
144         config: RepoConfig,
145         table: Union[FeatureTable, FeatureView],
146         data: List[
147             Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]
148         ],
149         progress: Optional[Callable[[int], Any]],
150     ) -> None:
151 
152         online_config = config.online_store
153         assert isinstance(online_config, DatastoreOnlineStoreConfig)
154         client = self._get_client(online_config)
155 
156         write_concurrency = online_config.write_concurrency
157         write_batch_size = online_config.write_batch_size
158         feast_project = config.project
159 
160         pool = ThreadPool(processes=write_concurrency)
161         pool.map(
162             lambda b: self._write_minibatch(client, feast_project, table, b, progress),
163             self._to_minibatches(data, batch_size=write_batch_size),
164         )
165 
166     @staticmethod
167     def _to_minibatches(data: ProtoBatch, batch_size) -> Iterator[ProtoBatch]:
168         """
169         Split data into minibatches, making sure we stay under GCP datastore transaction size
170         limits.
171         """
172         iterable = iter(data)
173 
174         while True:
175             batch = list(itertools.islice(iterable, batch_size))
176             if len(batch) > 0:
177                 yield batch
178             else:
179                 break
180 
181     @staticmethod
182     def _write_minibatch(
183         client,
184         project: str,
185         table: Union[FeatureTable, FeatureView],
186         data: Sequence[
187             Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]
188         ],
189         progress: Optional[Callable[[int], Any]],
190     ):
191         entities = []
192         for entity_key, features, timestamp, created_ts in data:
193             document_id = compute_entity_id(entity_key)
194 
195             key = client.key(
196                 "Project", project, "Table", table.name, "Row", document_id,
197             )
198 
199             entity = datastore.Entity(
200                 key=key, exclude_from_indexes=("created_ts", "event_ts", "values")
201             )
202 
203             entity.update(
204                 dict(
205                     key=entity_key.SerializeToString(),
206                     values={k: v.SerializeToString() for k, v in features.items()},
207                     event_ts=utils.make_tzaware(timestamp),
208                     created_ts=(
209                         utils.make_tzaware(created_ts)
210                         if created_ts is not None
211                         else None
212                     ),
213                 )
214             )
215             entities.append(entity)
216         with client.transaction():
217             client.put_multi(entities)
218 
219         if progress:
220             progress(len(entities))
221 
222     def online_read(
223         self,
224         config: RepoConfig,
225         table: Union[FeatureTable, FeatureView],
226         entity_keys: List[EntityKeyProto],
227         requested_features: Optional[List[str]] = None,
228     ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:
229 
230         online_config = config.online_store
231         assert isinstance(online_config, DatastoreOnlineStoreConfig)
232         client = self._get_client(online_config)
233 
234         feast_project = config.project
235 
236         result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []
237         for entity_key in entity_keys:
238             document_id = compute_entity_id(entity_key)
239             key = client.key(
240                 "Project", feast_project, "Table", table.name, "Row", document_id
241             )
242             value = client.get(key)
243             if value is not None:
244                 res = {}
245                 for feature_name, value_bin in value["values"].items():
246                     val = ValueProto()
247                     val.ParseFromString(value_bin)
248                     res[feature_name] = val
249                 result.append((value["event_ts"], res))
250             else:
251                 result.append((None, None))
252         return result
253 
254 
255 def _delete_all_values(client, key) -> None:
256     """
257     Delete all data under the key path in datastore.
258     """
259     while True:
260         query = client.query(kind="Row", ancestor=key)
261         entities = list(query.fetch(limit=1000))
262         if not entities:
263             return
264 
265         for entity in entities:
266             client.delete(entity.key)
267
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/feast/infra/online_stores/datastore.py b/sdk/python/feast/infra/online_stores/datastore.py
--- a/sdk/python/feast/infra/online_stores/datastore.py
+++ b/sdk/python/feast/infra/online_stores/datastore.py
@@ -30,6 +30,7 @@
 try:
     from google.auth.exceptions import DefaultCredentialsError
     from google.cloud import datastore
+    from google.cloud.datastore.client import Key
 except ImportError as e:
     from feast.errors import FeastExtrasDependencyImportError, FeastProviderLoginError
 
@@ -233,22 +234,32 @@
 
         feast_project = config.project
 
+        keys: List[Key] = []
         result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []
         for entity_key in entity_keys:
             document_id = compute_entity_id(entity_key)
             key = client.key(
                 "Project", feast_project, "Table", table.name, "Row", document_id
             )
-            value = client.get(key)
-            if value is not None:
+            keys.append(key)
+
+        values = client.get_multi(keys)
+
+        if values is not None:
+            keys_missing_from_response = set(keys) - set([v.key for v in values])
+            values = sorted(values, key=lambda v: keys.index(v.key))
+            for value in values:
                 res = {}
                 for feature_name, value_bin in value["values"].items():
                     val = ValueProto()
                     val.ParseFromString(value_bin)
                     res[feature_name] = val
                 result.append((value["event_ts"], res))
-            else:
-                result.append((None, None))
+            for missing_key_idx in sorted(
+                [keys.index(k) for k in keys_missing_from_response]
+            ):
+                result.insert(missing_key_idx, (None, None))
+
         return result

{"golden_diff": "diff --git a/sdk/python/feast/infra/online_stores/datastore.py b/sdk/python/feast/infra/online_stores/datastore.py\n--- a/sdk/python/feast/infra/online_stores/datastore.py\n+++ b/sdk/python/feast/infra/online_stores/datastore.py\n@@ -30,6 +30,7 @@\n try:\n     from google.auth.exceptions import DefaultCredentialsError\n     from google.cloud import datastore\n+    from google.cloud.datastore.client import Key\n except ImportError as e:\n     from feast.errors import FeastExtrasDependencyImportError, FeastProviderLoginError\n \n@@ -233,22 +234,32 @@\n \n         feast_project = config.project\n \n+        keys: List[Key] = []\n         result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []\n         for entity_key in entity_keys:\n             document_id = compute_entity_id(entity_key)\n             key = client.key(\n                 \"Project\", feast_project, \"Table\", table.name, \"Row\", document_id\n             )\n-            value = client.get(key)\n-            if value is not None:\n+            keys.append(key)\n+\n+        values = client.get_multi(keys)\n+\n+        if values is not None:\n+            keys_missing_from_response = set(keys) - set([v.key for v in values])\n+            values = sorted(values, key=lambda v: keys.index(v.key))\n+            for value in values:\n                 res = {}\n                 for feature_name, value_bin in value[\"values\"].items():\n                     val = ValueProto()\n                     val.ParseFromString(value_bin)\n                     res[feature_name] = val\n                 result.append((value[\"event_ts\"], res))\n-            else:\n-                result.append((None, None))\n+            for missing_key_idx in sorted(\n+                [keys.index(k) for k in keys_missing_from_response]\n+            ):\n+                result.insert(missing_key_idx, (None, None))\n+\n         return result\n", "issue": "Datastore online request makes a call once for each entity\n**Is your feature request related to a problem? Please describe.**\r\nThere are performance issues with Datastore when fetching features for multiple entities. An individual fetch request is made for each entity \r\n\r\n**Describe the solution you'd like**\r\nI would like to have a more efficient access to the Datastore for multiple entities. The python api for datastore offers a method `get_multi()` which allows to fetch multiple keys in one request.\r\n\r\n**Describe alternatives you've considered**\r\n---\r\n\r\n**Additional context**\r\n\r\nI have done some basic tests using the `get_multi` vs get and it seems that a improvement can be made by using the function `get_multi`\r\n\r\nHere is part of code  I used for testing:\r\n\r\n```\r\nimport numpy as np\r\nimport timeit\r\nfrom google.cloud import datastore\r\n\r\ndef multi_online_read(num_keys=5):\r\n\r\n    client = datastore.Client(\r\n                    project=project_name, namespace=None,\r\n                )\r\n    \r\n    feast_project = \"feature_repo\"\r\n    key = client.key(\"Project\", feast_project, \"Table\", table, \"Row\",  row)\r\n    keys = []\r\n    response = []\r\n    for i in range(num_keys):\r\n        keys.append(key)\r\n\r\n    response = client.get_multi(keys)\r\n\r\n    return response\r\n\r\ndef online_read(num_keys=5):\r\n    client = datastore.Client(\r\n                    project=project_name, namespace=None,\r\n                )\r\n    \r\n    feast_project = \"feature_repo\"\r\n    key = client.key(\"Project\", feast_project, \"Table\", table, \"Row\", row)\r\n    \r\n    response = []\r\n    for i in range(num_keys):\r\n        response.append(client.get(key))\r\n\r\n\r\n    return response\r\n\r\nif __name__ == \"__main__\":\r\n\r\n    results = timeit.repeat(online_read, repeat=10, number=1)\r\n    print(results)\r\n    print(np.mean(results))\r\n\r\n    results = timeit.repeat(multi_online_read, repeat=10, number=1)\r\n    print(results)\r\n    print(np.mean(results))\r\n```\r\n\r\n\r\n\r\n\r\n\n", "before_files": [{"content": "# Copyright 2021 The Feast Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     https://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport itertools\nfrom datetime import datetime\nfrom multiprocessing.pool import ThreadPool\nfrom typing import Any, Callable, Dict, Iterator, List, Optional, Sequence, Tuple, Union\n\nfrom pydantic import PositiveInt, StrictStr\nfrom pydantic.typing import Literal\n\nfrom feast import Entity, FeatureTable, utils\nfrom feast.feature_view import FeatureView\nfrom feast.infra.online_stores.helpers import compute_entity_id\nfrom feast.infra.online_stores.online_store import OnlineStore\nfrom feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto\nfrom feast.protos.feast.types.Value_pb2 import Value as ValueProto\nfrom feast.repo_config import FeastConfigBaseModel, RepoConfig\n\ntry:\n    from google.auth.exceptions import DefaultCredentialsError\n    from google.cloud import datastore\nexcept ImportError as e:\n    from feast.errors import FeastExtrasDependencyImportError, FeastProviderLoginError\n\n    raise FeastExtrasDependencyImportError(\"gcp\", str(e))\n\n\nProtoBatch = Sequence[\n    Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n]\n\n\nclass DatastoreOnlineStoreConfig(FeastConfigBaseModel):\n    \"\"\" Online store config for GCP Datastore \"\"\"\n\n    type: Literal[\"datastore\"] = \"datastore\"\n    \"\"\" Online store type selector\"\"\"\n\n    project_id: Optional[StrictStr] = None\n    \"\"\" (optional) GCP Project Id \"\"\"\n\n    namespace: Optional[StrictStr] = None\n    \"\"\" (optional) Datastore namespace \"\"\"\n\n    write_concurrency: Optional[PositiveInt] = 40\n    \"\"\" (optional) Amount of threads to use when writing batches of feature rows into Datastore\"\"\"\n\n    write_batch_size: Optional[PositiveInt] = 50\n    \"\"\" (optional) Amount of feature rows per batch being written into Datastore\"\"\"\n\n\nclass DatastoreOnlineStore(OnlineStore):\n    \"\"\"\n    OnlineStore is an object used for all interaction between Feast and the service used for offline storage of\n    features.\n    \"\"\"\n\n    _client: Optional[datastore.Client] = None\n\n    def update(\n        self,\n        config: RepoConfig,\n        tables_to_delete: Sequence[Union[FeatureTable, FeatureView]],\n        tables_to_keep: Sequence[Union[FeatureTable, FeatureView]],\n        entities_to_delete: Sequence[Entity],\n        entities_to_keep: Sequence[Entity],\n        partial: bool,\n    ):\n        \"\"\"\n        \"\"\"\n        online_config = config.online_store\n        assert isinstance(online_config, DatastoreOnlineStoreConfig)\n        client = self._get_client(online_config)\n        feast_project = config.project\n\n        for table in tables_to_keep:\n            key = client.key(\"Project\", feast_project, \"Table\", table.name)\n            entity = datastore.Entity(\n                key=key, exclude_from_indexes=(\"created_ts\", \"event_ts\", \"values\")\n            )\n            entity.update({\"created_ts\": datetime.utcnow()})\n            client.put(entity)\n\n        for table in tables_to_delete:\n            _delete_all_values(\n                client, client.key(\"Project\", feast_project, \"Table\", table.name)\n            )\n\n            # Delete the table metadata datastore entity\n            key = client.key(\"Project\", feast_project, \"Table\", table.name)\n            client.delete(key)\n\n    def teardown(\n        self,\n        config: RepoConfig,\n        tables: Sequence[Union[FeatureTable, FeatureView]],\n        entities: Sequence[Entity],\n    ):\n        \"\"\"\n        There's currently no teardown done for Datastore.\n        \"\"\"\n        online_config = config.online_store\n        assert isinstance(online_config, DatastoreOnlineStoreConfig)\n        client = self._get_client(online_config)\n        feast_project = config.project\n\n        for table in tables:\n            _delete_all_values(\n                client, client.key(\"Project\", feast_project, \"Table\", table.name)\n            )\n\n            # Delete the table metadata datastore entity\n            key = client.key(\"Project\", feast_project, \"Table\", table.name)\n            client.delete(key)\n\n    def _get_client(self, online_config: DatastoreOnlineStoreConfig):\n\n        if not self._client:\n            try:\n                self._client = datastore.Client(\n                    project=online_config.project_id, namespace=online_config.namespace,\n                )\n            except DefaultCredentialsError as e:\n                raise FeastProviderLoginError(\n                    str(e)\n                    + '\\nIt may be necessary to run \"gcloud auth application-default login\" if you would like to use your '\n                    \"local Google Cloud account \"\n                )\n        return self._client\n\n    def online_write_batch(\n        self,\n        config: RepoConfig,\n        table: Union[FeatureTable, FeatureView],\n        data: List[\n            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n        ],\n        progress: Optional[Callable[[int], Any]],\n    ) -> None:\n\n        online_config = config.online_store\n        assert isinstance(online_config, DatastoreOnlineStoreConfig)\n        client = self._get_client(online_config)\n\n        write_concurrency = online_config.write_concurrency\n        write_batch_size = online_config.write_batch_size\n        feast_project = config.project\n\n        pool = ThreadPool(processes=write_concurrency)\n        pool.map(\n            lambda b: self._write_minibatch(client, feast_project, table, b, progress),\n            self._to_minibatches(data, batch_size=write_batch_size),\n        )\n\n    @staticmethod\n    def _to_minibatches(data: ProtoBatch, batch_size) -> Iterator[ProtoBatch]:\n        \"\"\"\n        Split data into minibatches, making sure we stay under GCP datastore transaction size\n        limits.\n        \"\"\"\n        iterable = iter(data)\n\n        while True:\n            batch = list(itertools.islice(iterable, batch_size))\n            if len(batch) > 0:\n                yield batch\n            else:\n                break\n\n    @staticmethod\n    def _write_minibatch(\n        client,\n        project: str,\n        table: Union[FeatureTable, FeatureView],\n        data: Sequence[\n            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n        ],\n        progress: Optional[Callable[[int], Any]],\n    ):\n        entities = []\n        for entity_key, features, timestamp, created_ts in data:\n            document_id = compute_entity_id(entity_key)\n\n            key = client.key(\n                \"Project\", project, \"Table\", table.name, \"Row\", document_id,\n            )\n\n            entity = datastore.Entity(\n                key=key, exclude_from_indexes=(\"created_ts\", \"event_ts\", \"values\")\n            )\n\n            entity.update(\n                dict(\n                    key=entity_key.SerializeToString(),\n                    values={k: v.SerializeToString() for k, v in features.items()},\n                    event_ts=utils.make_tzaware(timestamp),\n                    created_ts=(\n                        utils.make_tzaware(created_ts)\n                        if created_ts is not None\n                        else None\n                    ),\n                )\n            )\n            entities.append(entity)\n        with client.transaction():\n            client.put_multi(entities)\n\n        if progress:\n            progress(len(entities))\n\n    def online_read(\n        self,\n        config: RepoConfig,\n        table: Union[FeatureTable, FeatureView],\n        entity_keys: List[EntityKeyProto],\n        requested_features: Optional[List[str]] = None,\n    ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:\n\n        online_config = config.online_store\n        assert isinstance(online_config, DatastoreOnlineStoreConfig)\n        client = self._get_client(online_config)\n\n        feast_project = config.project\n\n        result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []\n        for entity_key in entity_keys:\n            document_id = compute_entity_id(entity_key)\n            key = client.key(\n                \"Project\", feast_project, \"Table\", table.name, \"Row\", document_id\n            )\n            value = client.get(key)\n            if value is not None:\n                res = {}\n                for feature_name, value_bin in value[\"values\"].items():\n                    val = ValueProto()\n                    val.ParseFromString(value_bin)\n                    res[feature_name] = val\n                result.append((value[\"event_ts\"], res))\n            else:\n                result.append((None, None))\n        return result\n\n\ndef _delete_all_values(client, key) -> None:\n    \"\"\"\n    Delete all data under the key path in datastore.\n    \"\"\"\n    while True:\n        query = client.query(kind=\"Row\", ancestor=key)\n        entities = list(query.fetch(limit=1000))\n        if not entities:\n            return\n\n        for entity in entities:\n            client.delete(entity.key)\n", "path": "sdk/python/feast/infra/online_stores/datastore.py"}], "after_files": [{"content": "# Copyright 2021 The Feast Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     https://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport itertools\nfrom datetime import datetime\nfrom multiprocessing.pool import ThreadPool\nfrom typing import Any, Callable, Dict, Iterator, List, Optional, Sequence, Tuple, Union\n\nfrom pydantic import PositiveInt, StrictStr\nfrom pydantic.typing import Literal\n\nfrom feast import Entity, FeatureTable, utils\nfrom feast.feature_view import FeatureView\nfrom feast.infra.online_stores.helpers import compute_entity_id\nfrom feast.infra.online_stores.online_store import OnlineStore\nfrom feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto\nfrom feast.protos.feast.types.Value_pb2 import Value as ValueProto\nfrom feast.repo_config import FeastConfigBaseModel, RepoConfig\n\ntry:\n    from google.auth.exceptions import DefaultCredentialsError\n    from google.cloud import datastore\n    from google.cloud.datastore.client import Key\nexcept ImportError as e:\n    from feast.errors import FeastExtrasDependencyImportError, FeastProviderLoginError\n\n    raise FeastExtrasDependencyImportError(\"gcp\", str(e))\n\n\nProtoBatch = Sequence[\n    Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n]\n\n\nclass DatastoreOnlineStoreConfig(FeastConfigBaseModel):\n    \"\"\" Online store config for GCP Datastore \"\"\"\n\n    type: Literal[\"datastore\"] = \"datastore\"\n    \"\"\" Online store type selector\"\"\"\n\n    project_id: Optional[StrictStr] = None\n    \"\"\" (optional) GCP Project Id \"\"\"\n\n    namespace: Optional[StrictStr] = None\n    \"\"\" (optional) Datastore namespace \"\"\"\n\n    write_concurrency: Optional[PositiveInt] = 40\n    \"\"\" (optional) Amount of threads to use when writing batches of feature rows into Datastore\"\"\"\n\n    write_batch_size: Optional[PositiveInt] = 50\n    \"\"\" (optional) Amount of feature rows per batch being written into Datastore\"\"\"\n\n\nclass DatastoreOnlineStore(OnlineStore):\n    \"\"\"\n    OnlineStore is an object used for all interaction between Feast and the service used for offline storage of\n    features.\n    \"\"\"\n\n    _client: Optional[datastore.Client] = None\n\n    def update(\n        self,\n        config: RepoConfig,\n        tables_to_delete: Sequence[Union[FeatureTable, FeatureView]],\n        tables_to_keep: Sequence[Union[FeatureTable, FeatureView]],\n        entities_to_delete: Sequence[Entity],\n        entities_to_keep: Sequence[Entity],\n        partial: bool,\n    ):\n        \"\"\"\n        \"\"\"\n        online_config = config.online_store\n        assert isinstance(online_config, DatastoreOnlineStoreConfig)\n        client = self._get_client(online_config)\n        feast_project = config.project\n\n        for table in tables_to_keep:\n            key = client.key(\"Project\", feast_project, \"Table\", table.name)\n            entity = datastore.Entity(\n                key=key, exclude_from_indexes=(\"created_ts\", \"event_ts\", \"values\")\n            )\n            entity.update({\"created_ts\": datetime.utcnow()})\n            client.put(entity)\n\n        for table in tables_to_delete:\n            _delete_all_values(\n                client, client.key(\"Project\", feast_project, \"Table\", table.name)\n            )\n\n            # Delete the table metadata datastore entity\n            key = client.key(\"Project\", feast_project, \"Table\", table.name)\n            client.delete(key)\n\n    def teardown(\n        self,\n        config: RepoConfig,\n        tables: Sequence[Union[FeatureTable, FeatureView]],\n        entities: Sequence[Entity],\n    ):\n        \"\"\"\n        There's currently no teardown done for Datastore.\n        \"\"\"\n        online_config = config.online_store\n        assert isinstance(online_config, DatastoreOnlineStoreConfig)\n        client = self._get_client(online_config)\n        feast_project = config.project\n\n        for table in tables:\n            _delete_all_values(\n                client, client.key(\"Project\", feast_project, \"Table\", table.name)\n            )\n\n            # Delete the table metadata datastore entity\n            key = client.key(\"Project\", feast_project, \"Table\", table.name)\n            client.delete(key)\n\n    def _get_client(self, online_config: DatastoreOnlineStoreConfig):\n\n        if not self._client:\n            try:\n                self._client = datastore.Client(\n                    project=online_config.project_id, namespace=online_config.namespace,\n                )\n            except DefaultCredentialsError as e:\n                raise FeastProviderLoginError(\n                    str(e)\n                    + '\\nIt may be necessary to run \"gcloud auth application-default login\" if you would like to use your '\n                    \"local Google Cloud account \"\n                )\n        return self._client\n\n    def online_write_batch(\n        self,\n        config: RepoConfig,\n        table: Union[FeatureTable, FeatureView],\n        data: List[\n            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n        ],\n        progress: Optional[Callable[[int], Any]],\n    ) -> None:\n\n        online_config = config.online_store\n        assert isinstance(online_config, DatastoreOnlineStoreConfig)\n        client = self._get_client(online_config)\n\n        write_concurrency = online_config.write_concurrency\n        write_batch_size = online_config.write_batch_size\n        feast_project = config.project\n\n        pool = ThreadPool(processes=write_concurrency)\n        pool.map(\n            lambda b: self._write_minibatch(client, feast_project, table, b, progress),\n            self._to_minibatches(data, batch_size=write_batch_size),\n        )\n\n    @staticmethod\n    def _to_minibatches(data: ProtoBatch, batch_size) -> Iterator[ProtoBatch]:\n        \"\"\"\n        Split data into minibatches, making sure we stay under GCP datastore transaction size\n        limits.\n        \"\"\"\n        iterable = iter(data)\n\n        while True:\n            batch = list(itertools.islice(iterable, batch_size))\n            if len(batch) > 0:\n                yield batch\n            else:\n                break\n\n    @staticmethod\n    def _write_minibatch(\n        client,\n        project: str,\n        table: Union[FeatureTable, FeatureView],\n        data: Sequence[\n            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n        ],\n        progress: Optional[Callable[[int], Any]],\n    ):\n        entities = []\n        for entity_key, features, timestamp, created_ts in data:\n            document_id = compute_entity_id(entity_key)\n\n            key = client.key(\n                \"Project\", project, \"Table\", table.name, \"Row\", document_id,\n            )\n\n            entity = datastore.Entity(\n                key=key, exclude_from_indexes=(\"created_ts\", \"event_ts\", \"values\")\n            )\n\n            entity.update(\n                dict(\n                    key=entity_key.SerializeToString(),\n                    values={k: v.SerializeToString() for k, v in features.items()},\n                    event_ts=utils.make_tzaware(timestamp),\n                    created_ts=(\n                        utils.make_tzaware(created_ts)\n                        if created_ts is not None\n                        else None\n                    ),\n                )\n            )\n            entities.append(entity)\n        with client.transaction():\n            client.put_multi(entities)\n\n        if progress:\n            progress(len(entities))\n\n    def online_read(\n        self,\n        config: RepoConfig,\n        table: Union[FeatureTable, FeatureView],\n        entity_keys: List[EntityKeyProto],\n        requested_features: Optional[List[str]] = None,\n    ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:\n\n        online_config = config.online_store\n        assert isinstance(online_config, DatastoreOnlineStoreConfig)\n        client = self._get_client(online_config)\n\n        feast_project = config.project\n\n        keys: List[Key] = []\n        result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []\n        for entity_key in entity_keys:\n            document_id = compute_entity_id(entity_key)\n            key = client.key(\n                \"Project\", feast_project, \"Table\", table.name, \"Row\", document_id\n            )\n            keys.append(key)\n\n        values = client.get_multi(keys)\n\n        if values is not None:\n            keys_missing_from_response = set(keys) - set([v.key for v in values])\n            values = sorted(values, key=lambda v: keys.index(v.key))\n            for value in values:\n                res = {}\n                for feature_name, value_bin in value[\"values\"].items():\n                    val = ValueProto()\n                    val.ParseFromString(value_bin)\n                    res[feature_name] = val\n                result.append((value[\"event_ts\"], res))\n            for missing_key_idx in sorted(\n                [keys.index(k) for k in keys_missing_from_response]\n            ):\n                result.insert(missing_key_idx, (None, None))\n\n        return result\n\n\ndef _delete_all_values(client, key) -> None:\n    \"\"\"\n    Delete all data under the key path in datastore.\n    \"\"\"\n    while True:\n        query = client.query(kind=\"Row\", ancestor=key)\n        entities = list(query.fetch(limit=1000))\n        if not entities:\n            return\n\n        for entity in entities:\n            client.delete(entity.key)\n", "path": "sdk/python/feast/infra/online_stores/datastore.py"}]}