We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add delete_table to redis online store
Even if the feature view is no longer used, the data is not deleted until there are no feature views left in the join key.

Memory in redis is a very expensive resource.

When deleting feature_view, a function is added so that the corresponding key can be found and deleted in the redis hash.

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

{"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@@ -106,6 +106,39 @@\n \n         logger.debug(f\"Deleted {deleted_count} rows for entity {', '.join(join_keys)}\")\n \n+    def delete_table(self, config: RepoConfig, table: FeatureView):\n+        \"\"\"\n+        Delete all rows in Redis for a specific feature view\n+\n+        Args:\n+            config: Feast config\n+            table: Feature view to delete\n+        \"\"\"\n+        client = self._get_client(config.online_store)\n+        deleted_count = 0\n+        prefix = _redis_key_prefix(table.join_keys)\n+\n+        redis_hash_keys = [_mmh3(f\"{table.name}:{f.name}\") for f in table.features]\n+        redis_hash_keys.append(bytes(f\"_ts:{table.name}\", \"utf8\"))\n+\n+        with client.pipeline(transaction=False) as pipe:\n+            for _k in client.scan_iter(\n+                b\"\".join([prefix, b\"*\", config.project.encode(\"utf8\")])\n+            ):\n+                _tables = {\n+                    _hk[4:] for _hk in client.hgetall(_k) if _hk.startswith(b\"_ts:\")\n+                }\n+                if bytes(table.name, \"utf8\") not in _tables:\n+                    continue\n+                if len(_tables) == 1:\n+                    pipe.delete(_k)\n+                else:\n+                    pipe.hdel(_k, *redis_hash_keys)\n+                deleted_count += 1\n+            pipe.execute()\n+\n+        logger.debug(f\"Deleted {deleted_count} rows for feature view {table.name}\")\n+\n     @log_exceptions_and_usage(online_store=\"redis\")\n     def update(\n         self,\n@@ -117,16 +150,19 @@\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+        Delete data from feature views that are no longer in use.\n+\n+        Args:\n+            config: Feast config\n+            tables_to_delete: Feature views to delete\n+            tables_to_keep: Feature views to keep\n+            entities_to_delete: Entities to delete\n+            entities_to_keep: Entities to keep\n+            partial: Whether to do a partial update\n         \"\"\"\n-        join_keys_to_keep = set(tuple(table.join_keys) for table in tables_to_keep)\n \n-        join_keys_to_delete = set(tuple(table.join_keys) 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+        for table in tables_to_delete:\n+            self.delete_table(config, table)\n \n     def teardown(\n         self,\n", "issue": "Add delete_table to redis online store\nEven if the feature view is no longer used, the data is not deleted until there are no feature views left in the join key.\r\n\r\nMemory in redis is a very expensive resource.\r\n\r\nWhen deleting feature_view, a function is added so that the corresponding key can be found and deleted in the redis hash.\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\nimport pytz\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 ClusterNode, RedisCluster\n    from redis.sentinel import Sentinel\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    redis_sentinel = \"redis_sentinel\"\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    sentinel_master: StrictStr = \"mymaster\"\n    \"\"\"Sentinel's master name\"\"\"\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    key_ttl_seconds: Optional[int] = None\n    \"\"\"(Optional) redis key bin ttl (in seconds) for expiring entities\"\"\"\n\n\nclass RedisOnlineStore(OnlineStore):\n    \"\"\"\n    Redis implementation of the online store interface.\n\n    See https://github.com/feast-dev/feast/blob/master/docs/specs/online_store_format.md#redis-online-store-format\n    for more details about the data model for this implementation.\n\n    Attributes:\n        _client: Redis connection.\n    \"\"\"\n\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        prefix = _redis_key_prefix(join_keys)\n\n        with client.pipeline(transaction=False) as pipe:\n            for _k in client.scan_iter(\n                b\"\".join([prefix, b\"*\", config.project.encode(\"utf8\")])\n            ):\n                pipe.delete(_k)\n                deleted_count += 1\n            pipe.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.join_keys) for table in tables_to_keep)\n\n        join_keys_to_delete = set(tuple(table.join_keys) 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.join_keys) 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,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\"] = [\n                    ClusterNode(**node) for node in startup_nodes\n                ]\n                self._client = RedisCluster(**kwargs)\n            elif online_store_config.redis_type == RedisType.redis_sentinel:\n                sentinel_hosts = []\n\n                for item in startup_nodes:\n                    sentinel_hosts.append((item[\"host\"], int(item[\"port\"])))\n\n                sentinel = Sentinel(sentinel_hosts, **kwargs)\n                master = sentinel.master_for(online_store_config.sentinel_master)\n                self._client = master\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(transaction=False) 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(\n                    project,\n                    entity_key,\n                    entity_key_serialization_version=config.entity_key_serialization_version,\n                )\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\n                if online_store_config.key_ttl_seconds:\n                    pipe.expire(\n                        name=redis_key_bin, time=online_store_config.key_ttl_seconds\n                    )\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(\n                project,\n                entity_key,\n                entity_key_serialization_version=config.entity_key_serialization_version,\n            )\n            keys.append(redis_key_bin)\n        with client.pipeline(transaction=False) 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, tz=pytz.utc)\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\nimport pytz\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 ClusterNode, RedisCluster\n    from redis.sentinel import Sentinel\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    redis_sentinel = \"redis_sentinel\"\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    sentinel_master: StrictStr = \"mymaster\"\n    \"\"\"Sentinel's master name\"\"\"\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    key_ttl_seconds: Optional[int] = None\n    \"\"\"(Optional) redis key bin ttl (in seconds) for expiring entities\"\"\"\n\n\nclass RedisOnlineStore(OnlineStore):\n    \"\"\"\n    Redis implementation of the online store interface.\n\n    See https://github.com/feast-dev/feast/blob/master/docs/specs/online_store_format.md#redis-online-store-format\n    for more details about the data model for this implementation.\n\n    Attributes:\n        _client: Redis connection.\n    \"\"\"\n\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        prefix = _redis_key_prefix(join_keys)\n\n        with client.pipeline(transaction=False) as pipe:\n            for _k in client.scan_iter(\n                b\"\".join([prefix, b\"*\", config.project.encode(\"utf8\")])\n            ):\n                pipe.delete(_k)\n                deleted_count += 1\n            pipe.execute()\n\n        logger.debug(f\"Deleted {deleted_count} rows for entity {', '.join(join_keys)}\")\n\n    def delete_table(self, config: RepoConfig, table: FeatureView):\n        \"\"\"\n        Delete all rows in Redis for a specific feature view\n\n        Args:\n            config: Feast config\n            table: Feature view to delete\n        \"\"\"\n        client = self._get_client(config.online_store)\n        deleted_count = 0\n        prefix = _redis_key_prefix(table.join_keys)\n\n        redis_hash_keys = [_mmh3(f\"{table.name}:{f.name}\") for f in table.features]\n        redis_hash_keys.append(bytes(f\"_ts:{table.name}\", \"utf8\"))\n\n        with client.pipeline(transaction=False) as pipe:\n            for _k in client.scan_iter(\n                b\"\".join([prefix, b\"*\", config.project.encode(\"utf8\")])\n            ):\n                _tables = {\n                    _hk[4:] for _hk in client.hgetall(_k) if _hk.startswith(b\"_ts:\")\n                }\n                if bytes(table.name, \"utf8\") not in _tables:\n                    continue\n                if len(_tables) == 1:\n                    pipe.delete(_k)\n                else:\n                    pipe.hdel(_k, *redis_hash_keys)\n                deleted_count += 1\n            pipe.execute()\n\n        logger.debug(f\"Deleted {deleted_count} rows for feature view {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        Delete data from feature views that are no longer in use.\n\n        Args:\n            config: Feast config\n            tables_to_delete: Feature views to delete\n            tables_to_keep: Feature views to keep\n            entities_to_delete: Entities to delete\n            entities_to_keep: Entities to keep\n            partial: Whether to do a partial update\n        \"\"\"\n\n        for table in tables_to_delete:\n            self.delete_table(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        join_keys_to_delete = set(tuple(table.join_keys) 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,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\"] = [\n                    ClusterNode(**node) for node in startup_nodes\n                ]\n                self._client = RedisCluster(**kwargs)\n            elif online_store_config.redis_type == RedisType.redis_sentinel:\n                sentinel_hosts = []\n\n                for item in startup_nodes:\n                    sentinel_hosts.append((item[\"host\"], int(item[\"port\"])))\n\n                sentinel = Sentinel(sentinel_hosts, **kwargs)\n                master = sentinel.master_for(online_store_config.sentinel_master)\n                self._client = master\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(transaction=False) 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(\n                    project,\n                    entity_key,\n                    entity_key_serialization_version=config.entity_key_serialization_version,\n                )\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\n                if online_store_config.key_ttl_seconds:\n                    pipe.expire(\n                        name=redis_key_bin, time=online_store_config.key_ttl_seconds\n                    )\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(\n                project,\n                entity_key,\n                entity_key_serialization_version=config.entity_key_serialization_version,\n            )\n            keys.append(redis_key_bin)\n        with client.pipeline(transaction=False) 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, tz=pytz.utc)\n            return timestamp, res\n", "path": "sdk/python/feast/infra/online_stores/redis.py"}]}