We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support requested features on Postgresql online store
**Is your feature request related to a problem? Please describe.**
As a user, I want to retrieve subset of features on Postgresql online store instead of fetching all features

**Describe the solution you'd like**
Update online_read method to retrieve subset of features

**Additional context**
I'm close to creating a pull request for this issue myself.

--- 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/contrib/postgres.py`
Content:
```
1 import logging
2 from collections import defaultdict
3 from datetime import datetime
4 from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
5 
6 import psycopg2
7 import pytz
8 from psycopg2 import sql
9 from psycopg2.extras import execute_values
10 from pydantic.schema import Literal
11 
12 from feast import Entity
13 from feast.feature_view import FeatureView
14 from feast.infra.key_encoding_utils import serialize_entity_key
15 from feast.infra.online_stores.online_store import OnlineStore
16 from feast.infra.utils.postgres.connection_utils import _get_conn
17 from feast.infra.utils.postgres.postgres_config import PostgreSQLConfig
18 from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
19 from feast.protos.feast.types.Value_pb2 import Value as ValueProto
20 from feast.repo_config import RepoConfig
21 from feast.usage import log_exceptions_and_usage
22 
23 
24 class PostgreSQLOnlineStoreConfig(PostgreSQLConfig):
25     type: Literal["postgres"] = "postgres"
26 
27 
28 class PostgreSQLOnlineStore(OnlineStore):
29     _conn: Optional[psycopg2._psycopg.connection] = None
30 
31     def _get_conn(self, config: RepoConfig):
32         if not self._conn:
33             assert config.online_store.type == "postgres"
34             self._conn = _get_conn(config.online_store)
35         return self._conn
36 
37     @log_exceptions_and_usage(online_store="postgres")
38     def online_write_batch(
39         self,
40         config: RepoConfig,
41         table: FeatureView,
42         data: List[
43             Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]
44         ],
45         progress: Optional[Callable[[int], Any]],
46     ) -> None:
47         project = config.project
48 
49         with self._get_conn(config) as conn, conn.cursor() as cur:
50             insert_values = []
51             for entity_key, values, timestamp, created_ts in data:
52                 entity_key_bin = serialize_entity_key(
53                     entity_key,
54                     entity_key_serialization_version=config.entity_key_serialization_version,
55                 )
56                 timestamp = _to_naive_utc(timestamp)
57                 if created_ts is not None:
58                     created_ts = _to_naive_utc(created_ts)
59 
60                 for feature_name, val in values.items():
61                     insert_values.append(
62                         (
63                             entity_key_bin,
64                             feature_name,
65                             val.SerializeToString(),
66                             timestamp,
67                             created_ts,
68                         )
69                     )
70             # Control the batch so that we can update the progress
71             batch_size = 5000
72             for i in range(0, len(insert_values), batch_size):
73                 cur_batch = insert_values[i : i + batch_size]
74                 execute_values(
75                     cur,
76                     sql.SQL(
77                         """
78                         INSERT INTO {}
79                         (entity_key, feature_name, value, event_ts, created_ts)
80                         VALUES %s
81                         ON CONFLICT (entity_key, feature_name) DO
82                         UPDATE SET
83                             value = EXCLUDED.value,
84                             event_ts = EXCLUDED.event_ts,
85                             created_ts = EXCLUDED.created_ts;
86                         """,
87                     ).format(sql.Identifier(_table_id(project, table))),
88                     cur_batch,
89                     page_size=batch_size,
90                 )
91                 if progress:
92                     progress(len(cur_batch))
93 
94     @log_exceptions_and_usage(online_store="postgres")
95     def online_read(
96         self,
97         config: RepoConfig,
98         table: FeatureView,
99         entity_keys: List[EntityKeyProto],
100         requested_features: Optional[List[str]] = None,
101     ) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:
102         result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []
103 
104         project = config.project
105         with self._get_conn(config) as conn, conn.cursor() as cur:
106             # Collecting all the keys to a list allows us to make fewer round trips
107             # to PostgreSQL
108             keys = []
109             for entity_key in entity_keys:
110                 keys.append(
111                     serialize_entity_key(
112                         entity_key,
113                         entity_key_serialization_version=config.entity_key_serialization_version,
114                     )
115                 )
116 
117             cur.execute(
118                 sql.SQL(
119                     """
120                     SELECT entity_key, feature_name, value, event_ts
121                     FROM {} WHERE entity_key = ANY(%s);
122                     """
123                 ).format(
124                     sql.Identifier(_table_id(project, table)),
125                 ),
126                 (keys,),
127             )
128 
129             rows = cur.fetchall()
130 
131             # Since we don't know the order returned from PostgreSQL we'll need
132             # to construct a dict to be able to quickly look up the correct row
133             # when we iterate through the keys since they are in the correct order
134             values_dict = defaultdict(list)
135             for row in rows if rows is not None else []:
136                 values_dict[row[0].tobytes()].append(row[1:])
137 
138             for key in keys:
139                 if key in values_dict:
140                     value = values_dict[key]
141                     res = {}
142                     for feature_name, value_bin, event_ts in value:
143                         val = ValueProto()
144                         val.ParseFromString(bytes(value_bin))
145                         res[feature_name] = val
146                     result.append((event_ts, res))
147                 else:
148                     result.append((None, None))
149 
150         return result
151 
152     @log_exceptions_and_usage(online_store="postgres")
153     def update(
154         self,
155         config: RepoConfig,
156         tables_to_delete: Sequence[FeatureView],
157         tables_to_keep: Sequence[FeatureView],
158         entities_to_delete: Sequence[Entity],
159         entities_to_keep: Sequence[Entity],
160         partial: bool,
161     ):
162         project = config.project
163         schema_name = config.online_store.db_schema or config.online_store.user
164         with self._get_conn(config) as conn, conn.cursor() as cur:
165             # If a db_schema is provided, then that schema gets created if it doesn't
166             # exist. Else a schema is created for the feature store user.
167 
168             cur.execute(
169                 """
170                 SELECT schema_name
171                 FROM information_schema.schemata
172                 WHERE schema_name = %s
173                 """,
174                 (schema_name,),
175             )
176             schema_exists = cur.fetchone()
177             if not schema_exists:
178                 cur.execute(
179                     sql.SQL("CREATE SCHEMA IF NOT EXISTS {} AUTHORIZATION {}").format(
180                         sql.Identifier(schema_name),
181                         sql.Identifier(config.online_store.user),
182                     ),
183                 )
184 
185             for table in tables_to_delete:
186                 table_name = _table_id(project, table)
187                 cur.execute(_drop_table_and_index(table_name))
188 
189             for table in tables_to_keep:
190                 table_name = _table_id(project, table)
191                 cur.execute(
192                     sql.SQL(
193                         """
194                         CREATE TABLE IF NOT EXISTS {}
195                         (
196                             entity_key BYTEA,
197                             feature_name TEXT,
198                             value BYTEA,
199                             event_ts TIMESTAMPTZ,
200                             created_ts TIMESTAMPTZ,
201                             PRIMARY KEY(entity_key, feature_name)
202                         );
203                         CREATE INDEX IF NOT EXISTS {} ON {} (entity_key);
204                         """
205                     ).format(
206                         sql.Identifier(table_name),
207                         sql.Identifier(f"{table_name}_ek"),
208                         sql.Identifier(table_name),
209                     )
210                 )
211 
212             conn.commit()
213 
214     def teardown(
215         self,
216         config: RepoConfig,
217         tables: Sequence[FeatureView],
218         entities: Sequence[Entity],
219     ):
220         project = config.project
221         try:
222             with self._get_conn(config) as conn, conn.cursor() as cur:
223                 for table in tables:
224                     table_name = _table_id(project, table)
225                     cur.execute(_drop_table_and_index(table_name))
226         except Exception:
227             logging.exception("Teardown failed")
228             raise
229 
230 
231 def _table_id(project: str, table: FeatureView) -> str:
232     return f"{project}_{table.name}"
233 
234 
235 def _drop_table_and_index(table_name):
236     return sql.SQL(
237         """
238         DROP TABLE IF EXISTS {};
239         DROP INDEX IF EXISTS {};
240         """
241     ).format(
242         sql.Identifier(table_name),
243         sql.Identifier(f"{table_name}_ek"),
244     )
245 
246 
247 def _to_naive_utc(ts: datetime):
248     if ts.tzinfo is None:
249         return ts
250     else:
251         return ts.astimezone(pytz.utc).replace(tzinfo=None)
252
```
--- 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/contrib/postgres.py b/sdk/python/feast/infra/online_stores/contrib/postgres.py
--- a/sdk/python/feast/infra/online_stores/contrib/postgres.py
+++ b/sdk/python/feast/infra/online_stores/contrib/postgres.py
@@ -114,17 +114,30 @@
                     )
                 )
 
-            cur.execute(
-                sql.SQL(
-                    """
-                    SELECT entity_key, feature_name, value, event_ts
-                    FROM {} WHERE entity_key = ANY(%s);
-                    """
-                ).format(
-                    sql.Identifier(_table_id(project, table)),
-                ),
-                (keys,),
-            )
+            if not requested_features:
+                cur.execute(
+                    sql.SQL(
+                        """
+                        SELECT entity_key, feature_name, value, event_ts
+                        FROM {} WHERE entity_key = ANY(%s);
+                        """
+                    ).format(
+                        sql.Identifier(_table_id(project, table)),
+                    ),
+                    (keys,),
+                )
+            else:
+                cur.execute(
+                    sql.SQL(
+                        """
+                        SELECT entity_key, feature_name, value, event_ts
+                        FROM {} WHERE entity_key = ANY(%s) and feature_name = ANY(%s);
+                        """
+                    ).format(
+                        sql.Identifier(_table_id(project, table)),
+                    ),
+                    (keys, requested_features),
+                )
 
             rows = cur.fetchall()

{"golden_diff": "diff --git a/sdk/python/feast/infra/online_stores/contrib/postgres.py b/sdk/python/feast/infra/online_stores/contrib/postgres.py\n--- a/sdk/python/feast/infra/online_stores/contrib/postgres.py\n+++ b/sdk/python/feast/infra/online_stores/contrib/postgres.py\n@@ -114,17 +114,30 @@\n                     )\n                 )\n \n-            cur.execute(\n-                sql.SQL(\n-                    \"\"\"\n-                    SELECT entity_key, feature_name, value, event_ts\n-                    FROM {} WHERE entity_key = ANY(%s);\n-                    \"\"\"\n-                ).format(\n-                    sql.Identifier(_table_id(project, table)),\n-                ),\n-                (keys,),\n-            )\n+            if not requested_features:\n+                cur.execute(\n+                    sql.SQL(\n+                        \"\"\"\n+                        SELECT entity_key, feature_name, value, event_ts\n+                        FROM {} WHERE entity_key = ANY(%s);\n+                        \"\"\"\n+                    ).format(\n+                        sql.Identifier(_table_id(project, table)),\n+                    ),\n+                    (keys,),\n+                )\n+            else:\n+                cur.execute(\n+                    sql.SQL(\n+                        \"\"\"\n+                        SELECT entity_key, feature_name, value, event_ts\n+                        FROM {} WHERE entity_key = ANY(%s) and feature_name = ANY(%s);\n+                        \"\"\"\n+                    ).format(\n+                        sql.Identifier(_table_id(project, table)),\n+                    ),\n+                    (keys, requested_features),\n+                )\n \n             rows = cur.fetchall()\n", "issue": "Support requested features on Postgresql online store\n**Is your feature request related to a problem? Please describe.**\r\nAs a user, I want to retrieve subset of features on Postgresql online store instead of fetching all features\r\n\r\n**Describe the solution you'd like**\r\nUpdate online_read method to retrieve subset of features\r\n\r\n**Additional context**\r\nI'm close to creating a pull request for this issue myself.\n", "before_files": [{"content": "import logging\nfrom collections import defaultdict\nfrom datetime import datetime\nfrom typing import Any, Callable, Dict, List, Optional, Sequence, Tuple\n\nimport psycopg2\nimport pytz\nfrom psycopg2 import sql\nfrom psycopg2.extras import execute_values\nfrom pydantic.schema import Literal\n\nfrom feast import Entity\nfrom feast.feature_view import FeatureView\nfrom feast.infra.key_encoding_utils import serialize_entity_key\nfrom feast.infra.online_stores.online_store import OnlineStore\nfrom feast.infra.utils.postgres.connection_utils import _get_conn\nfrom feast.infra.utils.postgres.postgres_config import PostgreSQLConfig\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 RepoConfig\nfrom feast.usage import log_exceptions_and_usage\n\n\nclass PostgreSQLOnlineStoreConfig(PostgreSQLConfig):\n    type: Literal[\"postgres\"] = \"postgres\"\n\n\nclass PostgreSQLOnlineStore(OnlineStore):\n    _conn: Optional[psycopg2._psycopg.connection] = None\n\n    def _get_conn(self, config: RepoConfig):\n        if not self._conn:\n            assert config.online_store.type == \"postgres\"\n            self._conn = _get_conn(config.online_store)\n        return self._conn\n\n    @log_exceptions_and_usage(online_store=\"postgres\")\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        project = config.project\n\n        with self._get_conn(config) as conn, conn.cursor() as cur:\n            insert_values = []\n            for entity_key, values, timestamp, created_ts in data:\n                entity_key_bin = serialize_entity_key(\n                    entity_key,\n                    entity_key_serialization_version=config.entity_key_serialization_version,\n                )\n                timestamp = _to_naive_utc(timestamp)\n                if created_ts is not None:\n                    created_ts = _to_naive_utc(created_ts)\n\n                for feature_name, val in values.items():\n                    insert_values.append(\n                        (\n                            entity_key_bin,\n                            feature_name,\n                            val.SerializeToString(),\n                            timestamp,\n                            created_ts,\n                        )\n                    )\n            # Control the batch so that we can update the progress\n            batch_size = 5000\n            for i in range(0, len(insert_values), batch_size):\n                cur_batch = insert_values[i : i + batch_size]\n                execute_values(\n                    cur,\n                    sql.SQL(\n                        \"\"\"\n                        INSERT INTO {}\n                        (entity_key, feature_name, value, event_ts, created_ts)\n                        VALUES %s\n                        ON CONFLICT (entity_key, feature_name) DO\n                        UPDATE SET\n                            value = EXCLUDED.value,\n                            event_ts = EXCLUDED.event_ts,\n                            created_ts = EXCLUDED.created_ts;\n                        \"\"\",\n                    ).format(sql.Identifier(_table_id(project, table))),\n                    cur_batch,\n                    page_size=batch_size,\n                )\n                if progress:\n                    progress(len(cur_batch))\n\n    @log_exceptions_and_usage(online_store=\"postgres\")\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        result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []\n\n        project = config.project\n        with self._get_conn(config) as conn, conn.cursor() as cur:\n            # Collecting all the keys to a list allows us to make fewer round trips\n            # to PostgreSQL\n            keys = []\n            for entity_key in entity_keys:\n                keys.append(\n                    serialize_entity_key(\n                        entity_key,\n                        entity_key_serialization_version=config.entity_key_serialization_version,\n                    )\n                )\n\n            cur.execute(\n                sql.SQL(\n                    \"\"\"\n                    SELECT entity_key, feature_name, value, event_ts\n                    FROM {} WHERE entity_key = ANY(%s);\n                    \"\"\"\n                ).format(\n                    sql.Identifier(_table_id(project, table)),\n                ),\n                (keys,),\n            )\n\n            rows = cur.fetchall()\n\n            # Since we don't know the order returned from PostgreSQL we'll need\n            # to construct a dict to be able to quickly look up the correct row\n            # when we iterate through the keys since they are in the correct order\n            values_dict = defaultdict(list)\n            for row in rows if rows is not None else []:\n                values_dict[row[0].tobytes()].append(row[1:])\n\n            for key in keys:\n                if key in values_dict:\n                    value = values_dict[key]\n                    res = {}\n                    for feature_name, value_bin, event_ts in value:\n                        val = ValueProto()\n                        val.ParseFromString(bytes(value_bin))\n                        res[feature_name] = val\n                    result.append((event_ts, res))\n                else:\n                    result.append((None, None))\n\n        return result\n\n    @log_exceptions_and_usage(online_store=\"postgres\")\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        project = config.project\n        schema_name = config.online_store.db_schema or config.online_store.user\n        with self._get_conn(config) as conn, conn.cursor() as cur:\n            # If a db_schema is provided, then that schema gets created if it doesn't\n            # exist. Else a schema is created for the feature store user.\n\n            cur.execute(\n                \"\"\"\n                SELECT schema_name\n                FROM information_schema.schemata\n                WHERE schema_name = %s\n                \"\"\",\n                (schema_name,),\n            )\n            schema_exists = cur.fetchone()\n            if not schema_exists:\n                cur.execute(\n                    sql.SQL(\"CREATE SCHEMA IF NOT EXISTS {} AUTHORIZATION {}\").format(\n                        sql.Identifier(schema_name),\n                        sql.Identifier(config.online_store.user),\n                    ),\n                )\n\n            for table in tables_to_delete:\n                table_name = _table_id(project, table)\n                cur.execute(_drop_table_and_index(table_name))\n\n            for table in tables_to_keep:\n                table_name = _table_id(project, table)\n                cur.execute(\n                    sql.SQL(\n                        \"\"\"\n                        CREATE TABLE IF NOT EXISTS {}\n                        (\n                            entity_key BYTEA,\n                            feature_name TEXT,\n                            value BYTEA,\n                            event_ts TIMESTAMPTZ,\n                            created_ts TIMESTAMPTZ,\n                            PRIMARY KEY(entity_key, feature_name)\n                        );\n                        CREATE INDEX IF NOT EXISTS {} ON {} (entity_key);\n                        \"\"\"\n                    ).format(\n                        sql.Identifier(table_name),\n                        sql.Identifier(f\"{table_name}_ek\"),\n                        sql.Identifier(table_name),\n                    )\n                )\n\n            conn.commit()\n\n    def teardown(\n        self,\n        config: RepoConfig,\n        tables: Sequence[FeatureView],\n        entities: Sequence[Entity],\n    ):\n        project = config.project\n        try:\n            with self._get_conn(config) as conn, conn.cursor() as cur:\n                for table in tables:\n                    table_name = _table_id(project, table)\n                    cur.execute(_drop_table_and_index(table_name))\n        except Exception:\n            logging.exception(\"Teardown failed\")\n            raise\n\n\ndef _table_id(project: str, table: FeatureView) -> str:\n    return f\"{project}_{table.name}\"\n\n\ndef _drop_table_and_index(table_name):\n    return sql.SQL(\n        \"\"\"\n        DROP TABLE IF EXISTS {};\n        DROP INDEX IF EXISTS {};\n        \"\"\"\n    ).format(\n        sql.Identifier(table_name),\n        sql.Identifier(f\"{table_name}_ek\"),\n    )\n\n\ndef _to_naive_utc(ts: datetime):\n    if ts.tzinfo is None:\n        return ts\n    else:\n        return ts.astimezone(pytz.utc).replace(tzinfo=None)\n", "path": "sdk/python/feast/infra/online_stores/contrib/postgres.py"}], "after_files": [{"content": "import logging\nfrom collections import defaultdict\nfrom datetime import datetime\nfrom typing import Any, Callable, Dict, List, Optional, Sequence, Tuple\n\nimport psycopg2\nimport pytz\nfrom psycopg2 import sql\nfrom psycopg2.extras import execute_values\nfrom pydantic.schema import Literal\n\nfrom feast import Entity\nfrom feast.feature_view import FeatureView\nfrom feast.infra.key_encoding_utils import serialize_entity_key\nfrom feast.infra.online_stores.online_store import OnlineStore\nfrom feast.infra.utils.postgres.connection_utils import _get_conn\nfrom feast.infra.utils.postgres.postgres_config import PostgreSQLConfig\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 RepoConfig\nfrom feast.usage import log_exceptions_and_usage\n\n\nclass PostgreSQLOnlineStoreConfig(PostgreSQLConfig):\n    type: Literal[\"postgres\"] = \"postgres\"\n\n\nclass PostgreSQLOnlineStore(OnlineStore):\n    _conn: Optional[psycopg2._psycopg.connection] = None\n\n    def _get_conn(self, config: RepoConfig):\n        if not self._conn:\n            assert config.online_store.type == \"postgres\"\n            self._conn = _get_conn(config.online_store)\n        return self._conn\n\n    @log_exceptions_and_usage(online_store=\"postgres\")\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        project = config.project\n\n        with self._get_conn(config) as conn, conn.cursor() as cur:\n            insert_values = []\n            for entity_key, values, timestamp, created_ts in data:\n                entity_key_bin = serialize_entity_key(\n                    entity_key,\n                    entity_key_serialization_version=config.entity_key_serialization_version,\n                )\n                timestamp = _to_naive_utc(timestamp)\n                if created_ts is not None:\n                    created_ts = _to_naive_utc(created_ts)\n\n                for feature_name, val in values.items():\n                    insert_values.append(\n                        (\n                            entity_key_bin,\n                            feature_name,\n                            val.SerializeToString(),\n                            timestamp,\n                            created_ts,\n                        )\n                    )\n            # Control the batch so that we can update the progress\n            batch_size = 5000\n            for i in range(0, len(insert_values), batch_size):\n                cur_batch = insert_values[i : i + batch_size]\n                execute_values(\n                    cur,\n                    sql.SQL(\n                        \"\"\"\n                        INSERT INTO {}\n                        (entity_key, feature_name, value, event_ts, created_ts)\n                        VALUES %s\n                        ON CONFLICT (entity_key, feature_name) DO\n                        UPDATE SET\n                            value = EXCLUDED.value,\n                            event_ts = EXCLUDED.event_ts,\n                            created_ts = EXCLUDED.created_ts;\n                        \"\"\",\n                    ).format(sql.Identifier(_table_id(project, table))),\n                    cur_batch,\n                    page_size=batch_size,\n                )\n                if progress:\n                    progress(len(cur_batch))\n\n    @log_exceptions_and_usage(online_store=\"postgres\")\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        result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []\n\n        project = config.project\n        with self._get_conn(config) as conn, conn.cursor() as cur:\n            # Collecting all the keys to a list allows us to make fewer round trips\n            # to PostgreSQL\n            keys = []\n            for entity_key in entity_keys:\n                keys.append(\n                    serialize_entity_key(\n                        entity_key,\n                        entity_key_serialization_version=config.entity_key_serialization_version,\n                    )\n                )\n\n            if not requested_features:\n                cur.execute(\n                    sql.SQL(\n                        \"\"\"\n                        SELECT entity_key, feature_name, value, event_ts\n                        FROM {} WHERE entity_key = ANY(%s);\n                        \"\"\"\n                    ).format(\n                        sql.Identifier(_table_id(project, table)),\n                    ),\n                    (keys,),\n                )\n            else:\n                cur.execute(\n                    sql.SQL(\n                        \"\"\"\n                        SELECT entity_key, feature_name, value, event_ts\n                        FROM {} WHERE entity_key = ANY(%s) and feature_name = ANY(%s);\n                        \"\"\"\n                    ).format(\n                        sql.Identifier(_table_id(project, table)),\n                    ),\n                    (keys, requested_features),\n                )\n\n            rows = cur.fetchall()\n\n            # Since we don't know the order returned from PostgreSQL we'll need\n            # to construct a dict to be able to quickly look up the correct row\n            # when we iterate through the keys since they are in the correct order\n            values_dict = defaultdict(list)\n            for row in rows if rows is not None else []:\n                values_dict[row[0].tobytes()].append(row[1:])\n\n            for key in keys:\n                if key in values_dict:\n                    value = values_dict[key]\n                    res = {}\n                    for feature_name, value_bin, event_ts in value:\n                        val = ValueProto()\n                        val.ParseFromString(bytes(value_bin))\n                        res[feature_name] = val\n                    result.append((event_ts, res))\n                else:\n                    result.append((None, None))\n\n        return result\n\n    @log_exceptions_and_usage(online_store=\"postgres\")\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        project = config.project\n        schema_name = config.online_store.db_schema or config.online_store.user\n        with self._get_conn(config) as conn, conn.cursor() as cur:\n            # If a db_schema is provided, then that schema gets created if it doesn't\n            # exist. Else a schema is created for the feature store user.\n\n            cur.execute(\n                \"\"\"\n                SELECT schema_name\n                FROM information_schema.schemata\n                WHERE schema_name = %s\n                \"\"\",\n                (schema_name,),\n            )\n            schema_exists = cur.fetchone()\n            if not schema_exists:\n                cur.execute(\n                    sql.SQL(\"CREATE SCHEMA IF NOT EXISTS {} AUTHORIZATION {}\").format(\n                        sql.Identifier(schema_name),\n                        sql.Identifier(config.online_store.user),\n                    ),\n                )\n\n            for table in tables_to_delete:\n                table_name = _table_id(project, table)\n                cur.execute(_drop_table_and_index(table_name))\n\n            for table in tables_to_keep:\n                table_name = _table_id(project, table)\n                cur.execute(\n                    sql.SQL(\n                        \"\"\"\n                        CREATE TABLE IF NOT EXISTS {}\n                        (\n                            entity_key BYTEA,\n                            feature_name TEXT,\n                            value BYTEA,\n                            event_ts TIMESTAMPTZ,\n                            created_ts TIMESTAMPTZ,\n                            PRIMARY KEY(entity_key, feature_name)\n                        );\n                        CREATE INDEX IF NOT EXISTS {} ON {} (entity_key);\n                        \"\"\"\n                    ).format(\n                        sql.Identifier(table_name),\n                        sql.Identifier(f\"{table_name}_ek\"),\n                        sql.Identifier(table_name),\n                    )\n                )\n\n            conn.commit()\n\n    def teardown(\n        self,\n        config: RepoConfig,\n        tables: Sequence[FeatureView],\n        entities: Sequence[Entity],\n    ):\n        project = config.project\n        try:\n            with self._get_conn(config) as conn, conn.cursor() as cur:\n                for table in tables:\n                    table_name = _table_id(project, table)\n                    cur.execute(_drop_table_and_index(table_name))\n        except Exception:\n            logging.exception(\"Teardown failed\")\n            raise\n\n\ndef _table_id(project: str, table: FeatureView) -> str:\n    return f\"{project}_{table.name}\"\n\n\ndef _drop_table_and_index(table_name):\n    return sql.SQL(\n        \"\"\"\n        DROP TABLE IF EXISTS {};\n        DROP INDEX IF EXISTS {};\n        \"\"\"\n    ).format(\n        sql.Identifier(table_name),\n        sql.Identifier(f\"{table_name}_ek\"),\n    )\n\n\ndef _to_naive_utc(ts: datetime):\n    if ts.tzinfo is None:\n        return ts\n    else:\n        return ts.astimezone(pytz.utc).replace(tzinfo=None)\n", "path": "sdk/python/feast/infra/online_stores/contrib/postgres.py"}]}