We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Redis can delete incorrect keys
I haven't tested this explicitly but from the data model + this function it looks to be as if two FeatureViews have the same Entities (but are different tables) and one view is deleted the online keys for the other view will also be deleted... This seems like a bit of a flaw either in the Redis data model or in the deletion mechanism. 

https://github.com/feast-dev/feast/blob/ec4165396f70ab20b42246b093f777dfcc9f5277/sdk/python/feast/infra/online_stores/redis.py#L75-L88

--- 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/redis.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 json
15 import logging
16 from datetime import datetime
17 from enum import Enum
18 from typing import (
19     Any,
20     ByteString,
21     Callable,
22     Dict,
23     List,
24     Optional,
25     Sequence,
26     Tuple,
27     Union,
28 )
29 
30 from google.protobuf.timestamp_pb2 import Timestamp
31 from pydantic import StrictStr
32 from pydantic.typing import Literal
33 
34 from feast import Entity, FeatureView, RepoConfig, utils
35 from feast.infra.online_stores.helpers import _mmh3, _redis_key, _redis_key_prefix
36 from feast.infra.online_stores.online_store import OnlineStore
37 from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
38 from feast.protos.feast.types.Value_pb2 import Value as ValueProto
39 from feast.repo_config import FeastConfigBaseModel
40 from feast.usage import log_exceptions_and_usage, tracing_span
41 
42 try:
43     from redis import Redis
44     from redis.cluster import RedisCluster
45 except ImportError as e:
46     from feast.errors import FeastExtrasDependencyImportError
47 
48     raise FeastExtrasDependencyImportError("redis", str(e))
49 
50 logger = logging.getLogger(__name__)
51 
52 
53 class RedisType(str, Enum):
54     redis = "redis"
55     redis_cluster = "redis_cluster"
56 
57 
58 class RedisOnlineStoreConfig(FeastConfigBaseModel):
59     """Online store config for Redis store"""
60 
61     type: Literal["redis"] = "redis"
62     """Online store type selector"""
63 
64     redis_type: RedisType = RedisType.redis
65     """Redis type: redis or redis_cluster"""
66 
67     connection_string: StrictStr = "localhost:6379"
68     """Connection string containing the host, port, and configuration parameters for Redis
69      format: host:port,parameter1,parameter2 eg. redis:6379,db=0 """
70 
71 
72 class RedisOnlineStore(OnlineStore):
73     _client: Optional[Union[Redis, RedisCluster]] = None
74 
75     def delete_table_values(self, config: RepoConfig, table: FeatureView):
76         client = self._get_client(config.online_store)
77         deleted_count = 0
78         pipeline = client.pipeline()
79         prefix = _redis_key_prefix(table.entities)
80 
81         for _k in client.scan_iter(
82             b"".join([prefix, b"*", config.project.encode("utf8")])
83         ):
84             pipeline.delete(_k)
85             deleted_count += 1
86         pipeline.execute()
87 
88         logger.debug(f"Deleted {deleted_count} keys for {table.name}")
89 
90     @log_exceptions_and_usage(online_store="redis")
91     def update(
92         self,
93         config: RepoConfig,
94         tables_to_delete: Sequence[FeatureView],
95         tables_to_keep: Sequence[FeatureView],
96         entities_to_delete: Sequence[Entity],
97         entities_to_keep: Sequence[Entity],
98         partial: bool,
99     ):
100         """
101         We delete the keys in redis for tables/views being removed.
102         """
103         for table in tables_to_delete:
104             self.delete_table_values(config, table)
105 
106     def teardown(
107         self,
108         config: RepoConfig,
109         tables: Sequence[FeatureView],
110         entities: Sequence[Entity],
111     ):
112         """
113         We delete the keys in redis for tables/views being removed.
114         """
115         for table in tables:
116             self.delete_table_values(config, table)
117 
118     @staticmethod
119     def _parse_connection_string(connection_string: str):
120         """
121         Reads Redis connections string using format
122         for RedisCluster:
123             redis1:6379,redis2:6379,decode_responses=true,skip_full_coverage_check=true,ssl=true,password=...
124         for Redis:
125             redis_master:6379,db=0,ssl=true,password=...
126         """
127         startup_nodes = [
128             dict(zip(["host", "port"], c.split(":")))
129             for c in connection_string.split(",")
130             if "=" not in c
131         ]
132         params = {}
133         for c in connection_string.split(","):
134             if "=" in c:
135                 kv = c.split("=", 1)
136                 try:
137                     kv[1] = json.loads(kv[1])
138                 except json.JSONDecodeError:
139                     ...
140 
141                 it = iter(kv)
142                 params.update(dict(zip(it, it)))
143 
144         return startup_nodes, params
145 
146     def _get_client(self, online_store_config: RedisOnlineStoreConfig):
147         """
148         Creates the Redis client RedisCluster or Redis depending on configuration
149         """
150         if not self._client:
151             startup_nodes, kwargs = self._parse_connection_string(
152                 online_store_config.connection_string
153             )
154             if online_store_config.redis_type == RedisType.redis_cluster:
155                 kwargs["startup_nodes"] = startup_nodes
156                 self._client = RedisCluster(**kwargs)
157             else:
158                 kwargs["host"] = startup_nodes[0]["host"]
159                 kwargs["port"] = startup_nodes[0]["port"]
160                 self._client = Redis(**kwargs)
161         return self._client
162 
163     @log_exceptions_and_usage(online_store="redis")
164     def online_write_batch(
165         self,
166         config: RepoConfig,
167         table: FeatureView,
168         data: List[
169             Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]
170         ],
171         progress: Optional[Callable[[int], Any]],
172     ) -> None:
173         online_store_config = config.online_store
174         assert isinstance(online_store_config, RedisOnlineStoreConfig)
175 
176         client = self._get_client(online_store_config)
177         project = config.project
178 
179         feature_view = table.name
180         ts_key = f"_ts:{feature_view}"
181         keys = []
182         # redis pipelining optimization: send multiple commands to redis server without waiting for every reply
183         with client.pipeline() as pipe:
184             # check if a previous record under the key bin exists
185             # TODO: investigate if check and set is a better approach rather than pulling all entity ts and then setting
186             # it may be significantly slower but avoids potential (rare) race conditions
187             for entity_key, _, _, _ in data:
188                 redis_key_bin = _redis_key(project, entity_key)
189                 keys.append(redis_key_bin)
190                 pipe.hmget(redis_key_bin, ts_key)
191             prev_event_timestamps = pipe.execute()
192             # flattening the list of lists. `hmget` does the lookup assuming a list of keys in the key bin
193             prev_event_timestamps = [i[0] for i in prev_event_timestamps]
194 
195             for redis_key_bin, prev_event_time, (_, values, timestamp, _) in zip(
196                 keys, prev_event_timestamps, data
197             ):
198                 event_time_seconds = int(utils.make_tzaware(timestamp).timestamp())
199 
200                 # ignore if event_timestamp is before the event features that are currently in the feature store
201                 if prev_event_time:
202                     prev_ts = Timestamp()
203                     prev_ts.ParseFromString(prev_event_time)
204                     if prev_ts.seconds and event_time_seconds <= prev_ts.seconds:
205                         # TODO: somehow signal that it's not overwriting the current record?
206                         if progress:
207                             progress(1)
208                         continue
209 
210                 ts = Timestamp()
211                 ts.seconds = event_time_seconds
212                 entity_hset = dict()
213                 entity_hset[ts_key] = ts.SerializeToString()
214 
215                 for feature_name, val in values.items():
216                     f_key = _mmh3(f"{feature_view}:{feature_name}")
217                     entity_hset[f_key] = val.SerializeToString()
218 
219                 pipe.hset(redis_key_bin, mapping=entity_hset)
220                 # TODO: support expiring the entity / features in Redis
221                 # otherwise entity features remain in redis until cleaned up in separate process
222                 # client.expire redis_key_bin based a ttl setting
223             results = pipe.execute()
224             if progress:
225                 progress(len(results))
226 
227     @log_exceptions_and_usage(online_store="redis")
228     def online_read(
229         self,
230         config: RepoConfig,
231         table: FeatureView,
232         entity_keys: List[EntityKeyProto],
233         requested_features: Optional[List[str]] = None,
234     ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:
235         online_store_config = config.online_store
236         assert isinstance(online_store_config, RedisOnlineStoreConfig)
237 
238         client = self._get_client(online_store_config)
239         feature_view = table.name
240         project = config.project
241 
242         result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []
243 
244         if not requested_features:
245             requested_features = [f.name for f in table.features]
246 
247         hset_keys = [_mmh3(f"{feature_view}:{k}") for k in requested_features]
248 
249         ts_key = f"_ts:{feature_view}"
250         hset_keys.append(ts_key)
251         requested_features.append(ts_key)
252 
253         keys = []
254         for entity_key in entity_keys:
255             redis_key_bin = _redis_key(project, entity_key)
256             keys.append(redis_key_bin)
257         with client.pipeline() as pipe:
258             for redis_key_bin in keys:
259                 pipe.hmget(redis_key_bin, hset_keys)
260             with tracing_span(name="remote_call"):
261                 redis_values = pipe.execute()
262         for values in redis_values:
263             features = self._get_features_for_entity(
264                 values, feature_view, requested_features
265             )
266             result.append(features)
267         return result
268 
269     def _get_features_for_entity(
270         self,
271         values: List[ByteString],
272         feature_view: str,
273         requested_features: List[str],
274     ) -> Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]:
275         res_val = dict(zip(requested_features, values))
276 
277         res_ts = Timestamp()
278         ts_val = res_val.pop(f"_ts:{feature_view}")
279         if ts_val:
280             res_ts.ParseFromString(bytes(ts_val))
281 
282         res = {}
283         for feature_name, val_bin in res_val.items():
284             val = ValueProto()
285             if val_bin:
286                 val.ParseFromString(bytes(val_bin))
287             res[feature_name] = val
288 
289         if not res:
290             return None, None
291         else:
292             timestamp = datetime.fromtimestamp(res_ts.seconds)
293             return timestamp, res
294
```
--- 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/redis.py b/sdk/python/feast/infra/online_stores/redis.py
--- a/sdk/python/feast/infra/online_stores/redis.py
+++ b/sdk/python/feast/infra/online_stores/redis.py
@@ -72,11 +72,11 @@
 class RedisOnlineStore(OnlineStore):
     _client: Optional[Union[Redis, RedisCluster]] = None
 
-    def delete_table_values(self, config: RepoConfig, table: FeatureView):
+    def delete_entity_values(self, config: RepoConfig, join_keys: List[str]):
         client = self._get_client(config.online_store)
         deleted_count = 0
         pipeline = client.pipeline()
-        prefix = _redis_key_prefix(table.entities)
+        prefix = _redis_key_prefix(join_keys)
 
         for _k in client.scan_iter(
             b"".join([prefix, b"*", config.project.encode("utf8")])
@@ -85,7 +85,7 @@
             deleted_count += 1
         pipeline.execute()
 
-        logger.debug(f"Deleted {deleted_count} keys for {table.name}")
+        logger.debug(f"Deleted {deleted_count} rows for entity {', '.join(join_keys)}")
 
     @log_exceptions_and_usage(online_store="redis")
     def update(
@@ -98,10 +98,16 @@
         partial: bool,
     ):
         """
-        We delete the keys in redis for tables/views being removed.
+        Look for join_keys (list of entities) that are not in use anymore
+        (usually this happens when the last feature view that was using specific compound key is deleted)
+        and remove all features attached to this "join_keys".
         """
-        for table in tables_to_delete:
-            self.delete_table_values(config, table)
+        join_keys_to_keep = set(tuple(table.entities) for table in tables_to_keep)
+
+        join_keys_to_delete = set(tuple(table.entities) for table in tables_to_delete)
+
+        for join_keys in join_keys_to_delete - join_keys_to_keep:
+            self.delete_entity_values(config, list(join_keys))
 
     def teardown(
         self,
@@ -112,8 +118,10 @@
         """
         We delete the keys in redis for tables/views being removed.
         """
-        for table in tables:
-            self.delete_table_values(config, table)
+        join_keys_to_delete = set(tuple(table.entities) for table in tables)
+
+        for join_keys in join_keys_to_delete:
+            self.delete_entity_values(config, list(join_keys))
 
     @staticmethod
     def _parse_connection_string(connection_string: str):

{"golden_diff": "diff --git a/sdk/python/feast/infra/online_stores/redis.py b/sdk/python/feast/infra/online_stores/redis.py\n--- a/sdk/python/feast/infra/online_stores/redis.py\n+++ b/sdk/python/feast/infra/online_stores/redis.py\n@@ -72,11 +72,11 @@\n class RedisOnlineStore(OnlineStore):\n     _client: Optional[Union[Redis, RedisCluster]] = None\n \n-    def delete_table_values(self, config: RepoConfig, table: FeatureView):\n+    def delete_entity_values(self, config: RepoConfig, join_keys: List[str]):\n         client = self._get_client(config.online_store)\n         deleted_count = 0\n         pipeline = client.pipeline()\n-        prefix = _redis_key_prefix(table.entities)\n+        prefix = _redis_key_prefix(join_keys)\n \n         for _k in client.scan_iter(\n             b\"\".join([prefix, b\"*\", config.project.encode(\"utf8\")])\n@@ -85,7 +85,7 @@\n             deleted_count += 1\n         pipeline.execute()\n \n-        logger.debug(f\"Deleted {deleted_count} keys for {table.name}\")\n+        logger.debug(f\"Deleted {deleted_count} rows for entity {', '.join(join_keys)}\")\n \n     @log_exceptions_and_usage(online_store=\"redis\")\n     def update(\n@@ -98,10 +98,16 @@\n         partial: bool,\n     ):\n         \"\"\"\n-        We delete the keys in redis for tables/views being removed.\n+        Look for join_keys (list of entities) that are not in use anymore\n+        (usually this happens when the last feature view that was using specific compound key is deleted)\n+        and remove all features attached to this \"join_keys\".\n         \"\"\"\n-        for table in tables_to_delete:\n-            self.delete_table_values(config, table)\n+        join_keys_to_keep = set(tuple(table.entities) for table in tables_to_keep)\n+\n+        join_keys_to_delete = set(tuple(table.entities) for table in tables_to_delete)\n+\n+        for join_keys in join_keys_to_delete - join_keys_to_keep:\n+            self.delete_entity_values(config, list(join_keys))\n \n     def teardown(\n         self,\n@@ -112,8 +118,10 @@\n         \"\"\"\n         We delete the keys in redis for tables/views being removed.\n         \"\"\"\n-        for table in tables:\n-            self.delete_table_values(config, table)\n+        join_keys_to_delete = set(tuple(table.entities) for table in tables)\n+\n+        for join_keys in join_keys_to_delete:\n+            self.delete_entity_values(config, list(join_keys))\n \n     @staticmethod\n     def _parse_connection_string(connection_string: str):\n", "issue": "Redis can delete incorrect keys\nI haven't tested this explicitly but from the data model + this function it looks to be as if two FeatureViews have the same Entities (but are different tables) and one view is deleted the online keys for the other view will also be deleted... This seems like a bit of a flaw either in the Redis data model or in the deletion mechanism. \r\n\r\nhttps://github.com/feast-dev/feast/blob/ec4165396f70ab20b42246b093f777dfcc9f5277/sdk/python/feast/infra/online_stores/redis.py#L75-L88\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 json\nimport logging\nfrom datetime import datetime\nfrom enum import Enum\nfrom typing import (\n    Any,\n    ByteString,\n    Callable,\n    Dict,\n    List,\n    Optional,\n    Sequence,\n    Tuple,\n    Union,\n)\n\nfrom google.protobuf.timestamp_pb2 import Timestamp\nfrom pydantic import StrictStr\nfrom pydantic.typing import Literal\n\nfrom feast import Entity, FeatureView, RepoConfig, utils\nfrom feast.infra.online_stores.helpers import _mmh3, _redis_key, _redis_key_prefix\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\nfrom feast.usage import log_exceptions_and_usage, tracing_span\n\ntry:\n    from redis import Redis\n    from redis.cluster import RedisCluster\nexcept ImportError as e:\n    from feast.errors import FeastExtrasDependencyImportError\n\n    raise FeastExtrasDependencyImportError(\"redis\", str(e))\n\nlogger = logging.getLogger(__name__)\n\n\nclass RedisType(str, Enum):\n    redis = \"redis\"\n    redis_cluster = \"redis_cluster\"\n\n\nclass RedisOnlineStoreConfig(FeastConfigBaseModel):\n    \"\"\"Online store config for Redis store\"\"\"\n\n    type: Literal[\"redis\"] = \"redis\"\n    \"\"\"Online store type selector\"\"\"\n\n    redis_type: RedisType = RedisType.redis\n    \"\"\"Redis type: redis or redis_cluster\"\"\"\n\n    connection_string: StrictStr = \"localhost:6379\"\n    \"\"\"Connection string containing the host, port, and configuration parameters for Redis\n     format: host:port,parameter1,parameter2 eg. redis:6379,db=0 \"\"\"\n\n\nclass RedisOnlineStore(OnlineStore):\n    _client: Optional[Union[Redis, RedisCluster]] = None\n\n    def delete_table_values(self, config: RepoConfig, table: FeatureView):\n        client = self._get_client(config.online_store)\n        deleted_count = 0\n        pipeline = client.pipeline()\n        prefix = _redis_key_prefix(table.entities)\n\n        for _k in client.scan_iter(\n            b\"\".join([prefix, b\"*\", config.project.encode(\"utf8\")])\n        ):\n            pipeline.delete(_k)\n            deleted_count += 1\n        pipeline.execute()\n\n        logger.debug(f\"Deleted {deleted_count} keys for {table.name}\")\n\n    @log_exceptions_and_usage(online_store=\"redis\")\n    def update(\n        self,\n        config: RepoConfig,\n        tables_to_delete: Sequence[FeatureView],\n        tables_to_keep: Sequence[FeatureView],\n        entities_to_delete: Sequence[Entity],\n        entities_to_keep: Sequence[Entity],\n        partial: bool,\n    ):\n        \"\"\"\n        We delete the keys in redis for tables/views being removed.\n        \"\"\"\n        for table in tables_to_delete:\n            self.delete_table_values(config, table)\n\n    def teardown(\n        self,\n        config: RepoConfig,\n        tables: Sequence[FeatureView],\n        entities: Sequence[Entity],\n    ):\n        \"\"\"\n        We delete the keys in redis for tables/views being removed.\n        \"\"\"\n        for table in tables:\n            self.delete_table_values(config, table)\n\n    @staticmethod\n    def _parse_connection_string(connection_string: str):\n        \"\"\"\n        Reads Redis connections string using format\n        for RedisCluster:\n            redis1:6379,redis2:6379,decode_responses=true,skip_full_coverage_check=true,ssl=true,password=...\n        for Redis:\n            redis_master:6379,db=0,ssl=true,password=...\n        \"\"\"\n        startup_nodes = [\n            dict(zip([\"host\", \"port\"], c.split(\":\")))\n            for c in connection_string.split(\",\")\n            if \"=\" not in c\n        ]\n        params = {}\n        for c in connection_string.split(\",\"):\n            if \"=\" in c:\n                kv = c.split(\"=\", 1)\n                try:\n                    kv[1] = json.loads(kv[1])\n                except json.JSONDecodeError:\n                    ...\n\n                it = iter(kv)\n                params.update(dict(zip(it, it)))\n\n        return startup_nodes, params\n\n    def _get_client(self, online_store_config: RedisOnlineStoreConfig):\n        \"\"\"\n        Creates the Redis client RedisCluster or Redis depending on configuration\n        \"\"\"\n        if not self._client:\n            startup_nodes, kwargs = self._parse_connection_string(\n                online_store_config.connection_string\n            )\n            if online_store_config.redis_type == RedisType.redis_cluster:\n                kwargs[\"startup_nodes\"] = startup_nodes\n                self._client = RedisCluster(**kwargs)\n            else:\n                kwargs[\"host\"] = startup_nodes[0][\"host\"]\n                kwargs[\"port\"] = startup_nodes[0][\"port\"]\n                self._client = Redis(**kwargs)\n        return self._client\n\n    @log_exceptions_and_usage(online_store=\"redis\")\n    def online_write_batch(\n        self,\n        config: RepoConfig,\n        table: FeatureView,\n        data: List[\n            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n        ],\n        progress: Optional[Callable[[int], Any]],\n    ) -> None:\n        online_store_config = config.online_store\n        assert isinstance(online_store_config, RedisOnlineStoreConfig)\n\n        client = self._get_client(online_store_config)\n        project = config.project\n\n        feature_view = table.name\n        ts_key = f\"_ts:{feature_view}\"\n        keys = []\n        # redis pipelining optimization: send multiple commands to redis server without waiting for every reply\n        with client.pipeline() as pipe:\n            # check if a previous record under the key bin exists\n            # TODO: investigate if check and set is a better approach rather than pulling all entity ts and then setting\n            # it may be significantly slower but avoids potential (rare) race conditions\n            for entity_key, _, _, _ in data:\n                redis_key_bin = _redis_key(project, entity_key)\n                keys.append(redis_key_bin)\n                pipe.hmget(redis_key_bin, ts_key)\n            prev_event_timestamps = pipe.execute()\n            # flattening the list of lists. `hmget` does the lookup assuming a list of keys in the key bin\n            prev_event_timestamps = [i[0] for i in prev_event_timestamps]\n\n            for redis_key_bin, prev_event_time, (_, values, timestamp, _) in zip(\n                keys, prev_event_timestamps, data\n            ):\n                event_time_seconds = int(utils.make_tzaware(timestamp).timestamp())\n\n                # ignore if event_timestamp is before the event features that are currently in the feature store\n                if prev_event_time:\n                    prev_ts = Timestamp()\n                    prev_ts.ParseFromString(prev_event_time)\n                    if prev_ts.seconds and event_time_seconds <= prev_ts.seconds:\n                        # TODO: somehow signal that it's not overwriting the current record?\n                        if progress:\n                            progress(1)\n                        continue\n\n                ts = Timestamp()\n                ts.seconds = event_time_seconds\n                entity_hset = dict()\n                entity_hset[ts_key] = ts.SerializeToString()\n\n                for feature_name, val in values.items():\n                    f_key = _mmh3(f\"{feature_view}:{feature_name}\")\n                    entity_hset[f_key] = val.SerializeToString()\n\n                pipe.hset(redis_key_bin, mapping=entity_hset)\n                # TODO: support expiring the entity / features in Redis\n                # otherwise entity features remain in redis until cleaned up in separate process\n                # client.expire redis_key_bin based a ttl setting\n            results = pipe.execute()\n            if progress:\n                progress(len(results))\n\n    @log_exceptions_and_usage(online_store=\"redis\")\n    def online_read(\n        self,\n        config: RepoConfig,\n        table: FeatureView,\n        entity_keys: List[EntityKeyProto],\n        requested_features: Optional[List[str]] = None,\n    ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:\n        online_store_config = config.online_store\n        assert isinstance(online_store_config, RedisOnlineStoreConfig)\n\n        client = self._get_client(online_store_config)\n        feature_view = table.name\n        project = config.project\n\n        result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []\n\n        if not requested_features:\n            requested_features = [f.name for f in table.features]\n\n        hset_keys = [_mmh3(f\"{feature_view}:{k}\") for k in requested_features]\n\n        ts_key = f\"_ts:{feature_view}\"\n        hset_keys.append(ts_key)\n        requested_features.append(ts_key)\n\n        keys = []\n        for entity_key in entity_keys:\n            redis_key_bin = _redis_key(project, entity_key)\n            keys.append(redis_key_bin)\n        with client.pipeline() as pipe:\n            for redis_key_bin in keys:\n                pipe.hmget(redis_key_bin, hset_keys)\n            with tracing_span(name=\"remote_call\"):\n                redis_values = pipe.execute()\n        for values in redis_values:\n            features = self._get_features_for_entity(\n                values, feature_view, requested_features\n            )\n            result.append(features)\n        return result\n\n    def _get_features_for_entity(\n        self,\n        values: List[ByteString],\n        feature_view: str,\n        requested_features: List[str],\n    ) -> Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]:\n        res_val = dict(zip(requested_features, values))\n\n        res_ts = Timestamp()\n        ts_val = res_val.pop(f\"_ts:{feature_view}\")\n        if ts_val:\n            res_ts.ParseFromString(bytes(ts_val))\n\n        res = {}\n        for feature_name, val_bin in res_val.items():\n            val = ValueProto()\n            if val_bin:\n                val.ParseFromString(bytes(val_bin))\n            res[feature_name] = val\n\n        if not res:\n            return None, None\n        else:\n            timestamp = datetime.fromtimestamp(res_ts.seconds)\n            return timestamp, res\n", "path": "sdk/python/feast/infra/online_stores/redis.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 json\nimport logging\nfrom datetime import datetime\nfrom enum import Enum\nfrom typing import (\n    Any,\n    ByteString,\n    Callable,\n    Dict,\n    List,\n    Optional,\n    Sequence,\n    Tuple,\n    Union,\n)\n\nfrom google.protobuf.timestamp_pb2 import Timestamp\nfrom pydantic import StrictStr\nfrom pydantic.typing import Literal\n\nfrom feast import Entity, FeatureView, RepoConfig, utils\nfrom feast.infra.online_stores.helpers import _mmh3, _redis_key, _redis_key_prefix\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\nfrom feast.usage import log_exceptions_and_usage, tracing_span\n\ntry:\n    from redis import Redis\n    from redis.cluster import RedisCluster\nexcept ImportError as e:\n    from feast.errors import FeastExtrasDependencyImportError\n\n    raise FeastExtrasDependencyImportError(\"redis\", str(e))\n\nlogger = logging.getLogger(__name__)\n\n\nclass RedisType(str, Enum):\n    redis = \"redis\"\n    redis_cluster = \"redis_cluster\"\n\n\nclass RedisOnlineStoreConfig(FeastConfigBaseModel):\n    \"\"\"Online store config for Redis store\"\"\"\n\n    type: Literal[\"redis\"] = \"redis\"\n    \"\"\"Online store type selector\"\"\"\n\n    redis_type: RedisType = RedisType.redis\n    \"\"\"Redis type: redis or redis_cluster\"\"\"\n\n    connection_string: StrictStr = \"localhost:6379\"\n    \"\"\"Connection string containing the host, port, and configuration parameters for Redis\n     format: host:port,parameter1,parameter2 eg. redis:6379,db=0 \"\"\"\n\n\nclass RedisOnlineStore(OnlineStore):\n    _client: Optional[Union[Redis, RedisCluster]] = None\n\n    def delete_entity_values(self, config: RepoConfig, join_keys: List[str]):\n        client = self._get_client(config.online_store)\n        deleted_count = 0\n        pipeline = client.pipeline()\n        prefix = _redis_key_prefix(join_keys)\n\n        for _k in client.scan_iter(\n            b\"\".join([prefix, b\"*\", config.project.encode(\"utf8\")])\n        ):\n            pipeline.delete(_k)\n            deleted_count += 1\n        pipeline.execute()\n\n        logger.debug(f\"Deleted {deleted_count} rows for entity {', '.join(join_keys)}\")\n\n    @log_exceptions_and_usage(online_store=\"redis\")\n    def update(\n        self,\n        config: RepoConfig,\n        tables_to_delete: Sequence[FeatureView],\n        tables_to_keep: Sequence[FeatureView],\n        entities_to_delete: Sequence[Entity],\n        entities_to_keep: Sequence[Entity],\n        partial: bool,\n    ):\n        \"\"\"\n        Look for join_keys (list of entities) that are not in use anymore\n        (usually this happens when the last feature view that was using specific compound key is deleted)\n        and remove all features attached to this \"join_keys\".\n        \"\"\"\n        join_keys_to_keep = set(tuple(table.entities) for table in tables_to_keep)\n\n        join_keys_to_delete = set(tuple(table.entities) for table in tables_to_delete)\n\n        for join_keys in join_keys_to_delete - join_keys_to_keep:\n            self.delete_entity_values(config, list(join_keys))\n\n    def teardown(\n        self,\n        config: RepoConfig,\n        tables: Sequence[FeatureView],\n        entities: Sequence[Entity],\n    ):\n        \"\"\"\n        We delete the keys in redis for tables/views being removed.\n        \"\"\"\n        join_keys_to_delete = set(tuple(table.entities) for table in tables)\n\n        for join_keys in join_keys_to_delete:\n            self.delete_entity_values(config, list(join_keys))\n\n    @staticmethod\n    def _parse_connection_string(connection_string: str):\n        \"\"\"\n        Reads Redis connections string using format\n        for RedisCluster:\n            redis1:6379,redis2:6379,decode_responses=true,skip_full_coverage_check=true,ssl=true,password=...\n        for Redis:\n            redis_master:6379,db=0,ssl=true,password=...\n        \"\"\"\n        startup_nodes = [\n            dict(zip([\"host\", \"port\"], c.split(\":\")))\n            for c in connection_string.split(\",\")\n            if \"=\" not in c\n        ]\n        params = {}\n        for c in connection_string.split(\",\"):\n            if \"=\" in c:\n                kv = c.split(\"=\", 1)\n                try:\n                    kv[1] = json.loads(kv[1])\n                except json.JSONDecodeError:\n                    ...\n\n                it = iter(kv)\n                params.update(dict(zip(it, it)))\n\n        return startup_nodes, params\n\n    def _get_client(self, online_store_config: RedisOnlineStoreConfig):\n        \"\"\"\n        Creates the Redis client RedisCluster or Redis depending on configuration\n        \"\"\"\n        if not self._client:\n            startup_nodes, kwargs = self._parse_connection_string(\n                online_store_config.connection_string\n            )\n            if online_store_config.redis_type == RedisType.redis_cluster:\n                kwargs[\"startup_nodes\"] = startup_nodes\n                self._client = RedisCluster(**kwargs)\n            else:\n                kwargs[\"host\"] = startup_nodes[0][\"host\"]\n                kwargs[\"port\"] = startup_nodes[0][\"port\"]\n                self._client = Redis(**kwargs)\n        return self._client\n\n    @log_exceptions_and_usage(online_store=\"redis\")\n    def online_write_batch(\n        self,\n        config: RepoConfig,\n        table: FeatureView,\n        data: List[\n            Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]\n        ],\n        progress: Optional[Callable[[int], Any]],\n    ) -> None:\n        online_store_config = config.online_store\n        assert isinstance(online_store_config, RedisOnlineStoreConfig)\n\n        client = self._get_client(online_store_config)\n        project = config.project\n\n        feature_view = table.name\n        ts_key = f\"_ts:{feature_view}\"\n        keys = []\n        # redis pipelining optimization: send multiple commands to redis server without waiting for every reply\n        with client.pipeline() as pipe:\n            # check if a previous record under the key bin exists\n            # TODO: investigate if check and set is a better approach rather than pulling all entity ts and then setting\n            # it may be significantly slower but avoids potential (rare) race conditions\n            for entity_key, _, _, _ in data:\n                redis_key_bin = _redis_key(project, entity_key)\n                keys.append(redis_key_bin)\n                pipe.hmget(redis_key_bin, ts_key)\n            prev_event_timestamps = pipe.execute()\n            # flattening the list of lists. `hmget` does the lookup assuming a list of keys in the key bin\n            prev_event_timestamps = [i[0] for i in prev_event_timestamps]\n\n            for redis_key_bin, prev_event_time, (_, values, timestamp, _) in zip(\n                keys, prev_event_timestamps, data\n            ):\n                event_time_seconds = int(utils.make_tzaware(timestamp).timestamp())\n\n                # ignore if event_timestamp is before the event features that are currently in the feature store\n                if prev_event_time:\n                    prev_ts = Timestamp()\n                    prev_ts.ParseFromString(prev_event_time)\n                    if prev_ts.seconds and event_time_seconds <= prev_ts.seconds:\n                        # TODO: somehow signal that it's not overwriting the current record?\n                        if progress:\n                            progress(1)\n                        continue\n\n                ts = Timestamp()\n                ts.seconds = event_time_seconds\n                entity_hset = dict()\n                entity_hset[ts_key] = ts.SerializeToString()\n\n                for feature_name, val in values.items():\n                    f_key = _mmh3(f\"{feature_view}:{feature_name}\")\n                    entity_hset[f_key] = val.SerializeToString()\n\n                pipe.hset(redis_key_bin, mapping=entity_hset)\n                # TODO: support expiring the entity / features in Redis\n                # otherwise entity features remain in redis until cleaned up in separate process\n                # client.expire redis_key_bin based a ttl setting\n            results = pipe.execute()\n            if progress:\n                progress(len(results))\n\n    @log_exceptions_and_usage(online_store=\"redis\")\n    def online_read(\n        self,\n        config: RepoConfig,\n        table: FeatureView,\n        entity_keys: List[EntityKeyProto],\n        requested_features: Optional[List[str]] = None,\n    ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:\n        online_store_config = config.online_store\n        assert isinstance(online_store_config, RedisOnlineStoreConfig)\n\n        client = self._get_client(online_store_config)\n        feature_view = table.name\n        project = config.project\n\n        result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []\n\n        if not requested_features:\n            requested_features = [f.name for f in table.features]\n\n        hset_keys = [_mmh3(f\"{feature_view}:{k}\") for k in requested_features]\n\n        ts_key = f\"_ts:{feature_view}\"\n        hset_keys.append(ts_key)\n        requested_features.append(ts_key)\n\n        keys = []\n        for entity_key in entity_keys:\n            redis_key_bin = _redis_key(project, entity_key)\n            keys.append(redis_key_bin)\n        with client.pipeline() as pipe:\n            for redis_key_bin in keys:\n                pipe.hmget(redis_key_bin, hset_keys)\n            with tracing_span(name=\"remote_call\"):\n                redis_values = pipe.execute()\n        for values in redis_values:\n            features = self._get_features_for_entity(\n                values, feature_view, requested_features\n            )\n            result.append(features)\n        return result\n\n    def _get_features_for_entity(\n        self,\n        values: List[ByteString],\n        feature_view: str,\n        requested_features: List[str],\n    ) -> Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]:\n        res_val = dict(zip(requested_features, values))\n\n        res_ts = Timestamp()\n        ts_val = res_val.pop(f\"_ts:{feature_view}\")\n        if ts_val:\n            res_ts.ParseFromString(bytes(ts_val))\n\n        res = {}\n        for feature_name, val_bin in res_val.items():\n            val = ValueProto()\n            if val_bin:\n                val.ParseFromString(bytes(val_bin))\n            res[feature_name] = val\n\n        if not res:\n            return None, None\n        else:\n            timestamp = datetime.fromtimestamp(res_ts.seconds)\n            return timestamp, res\n", "path": "sdk/python/feast/infra/online_stores/redis.py"}]}