We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Feast push (Redshift/DynamoDb) not work with PushMode.ONLINE_AND_OFFLINE when more than 500 columns
## Expected Behavior 
Currently, we have a push source with Redshift Offline Store and DynamoDb  Online Store. 
We built our view with more than 500 columns. Around 750 columns. 

We expected to ingest data in dynamo and redshift when we run 
`fs.push("push_source", df, to=PushMode.ONLINE_AND_OFFLINE)`

## Current Behavior

Push command raise an issue like` [ERROR] ValueError: The input dataframe has columns ..`
This issue come from `get_table_column_names_and_types `method in `write_to_offline_store` method.
In the method, we check if `if set(input_columns) != set(source_columns)` and raise the below issue if there are diff. 

In case with more than 500 columns we get a diff because source_columns come from `get_table_column_names_and_types` method result where the result is define by MaxResults parameters. 

## Steps to reproduce
```
entity= Entity(
    name="entity",
    join_keys=["entity_id"],
    value_type=ValueType.INT64,
)

push_source = PushSource(
    name="push_source",
    batch_source=RedshiftSource(
        table="fs_push_view",
        timestamp_field="datecreation",
        created_timestamp_column="created_at"),
)

besoin_embedding_push_view = FeatureView(
    name="push_view",
    entities=[entity],
    schema=[Field(name=f"field_{dim}", dtype=types.Float64) for dim in range(768)],
    source=push_source 
)

fs.push("push_source", df, to=PushMode.ONLINE_AND_OFFLINE)

```

### Specifications

- Version: 0.25.0
- Platform: AWS
- Subsystem:

## Possible Solution

In my mind, we have two solutions:

- Set higher MaxResults in describe_table method
- Use NextToken to iterate through results



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/feast/infra/offline_stores/redshift_source.py`
Content:
```
1 from typing import Callable, Dict, Iterable, Optional, Tuple
2 
3 from typeguard import typechecked
4 
5 from feast import type_map
6 from feast.data_source import DataSource
7 from feast.errors import (
8     DataSourceNoNameException,
9     DataSourceNotFoundException,
10     RedshiftCredentialsError,
11 )
12 from feast.feature_logging import LoggingDestination
13 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
14 from feast.protos.feast.core.FeatureService_pb2 import (
15     LoggingConfig as LoggingConfigProto,
16 )
17 from feast.protos.feast.core.SavedDataset_pb2 import (
18     SavedDatasetStorage as SavedDatasetStorageProto,
19 )
20 from feast.repo_config import RepoConfig
21 from feast.saved_dataset import SavedDatasetStorage
22 from feast.value_type import ValueType
23 
24 
25 @typechecked
26 class RedshiftSource(DataSource):
27     def __init__(
28         self,
29         *,
30         name: Optional[str] = None,
31         timestamp_field: Optional[str] = "",
32         table: Optional[str] = None,
33         schema: Optional[str] = None,
34         created_timestamp_column: Optional[str] = "",
35         field_mapping: Optional[Dict[str, str]] = None,
36         query: Optional[str] = None,
37         description: Optional[str] = "",
38         tags: Optional[Dict[str, str]] = None,
39         owner: Optional[str] = "",
40         database: Optional[str] = "",
41     ):
42         """
43         Creates a RedshiftSource object.
44 
45         Args:
46             name (optional): Name for the source. Defaults to the table if not specified, in which
47                 case the table must be specified.
48             timestamp_field (optional): Event timestamp field used for point in time
49                 joins of feature values.
50             table (optional): Redshift table where the features are stored. Exactly one of 'table'
51                 and 'query' must be specified.
52             schema (optional): Redshift schema in which the table is located.
53             created_timestamp_column (optional): Timestamp column indicating when the
54                 row was created, used for deduplicating rows.
55             field_mapping (optional): A dictionary mapping of column names in this data
56                 source to column names in a feature table or view.
57             query (optional): The query to be executed to obtain the features. Exactly one of 'table'
58                 and 'query' must be specified.
59             description (optional): A human-readable description.
60             tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
61             owner (optional): The owner of the redshift source, typically the email of the primary
62                 maintainer.
63             database (optional): The Redshift database name.
64         """
65         if table is None and query is None:
66             raise ValueError('No "table" or "query" argument provided.')
67 
68         # The default Redshift schema is named "public".
69         _schema = "public" if table and not schema else schema
70         self.redshift_options = RedshiftOptions(
71             table=table, schema=_schema, query=query, database=database
72         )
73 
74         # If no name, use the table as the default name.
75         if name is None and table is None:
76             raise DataSourceNoNameException()
77         name = name or table
78         assert name
79 
80         super().__init__(
81             name=name,
82             timestamp_field=timestamp_field,
83             created_timestamp_column=created_timestamp_column,
84             field_mapping=field_mapping,
85             description=description,
86             tags=tags,
87             owner=owner,
88         )
89 
90     @staticmethod
91     def from_proto(data_source: DataSourceProto):
92         """
93         Creates a RedshiftSource from a protobuf representation of a RedshiftSource.
94 
95         Args:
96             data_source: A protobuf representation of a RedshiftSource
97 
98         Returns:
99             A RedshiftSource object based on the data_source protobuf.
100         """
101         return RedshiftSource(
102             name=data_source.name,
103             timestamp_field=data_source.timestamp_field,
104             table=data_source.redshift_options.table,
105             schema=data_source.redshift_options.schema,
106             created_timestamp_column=data_source.created_timestamp_column,
107             field_mapping=dict(data_source.field_mapping),
108             query=data_source.redshift_options.query,
109             description=data_source.description,
110             tags=dict(data_source.tags),
111             owner=data_source.owner,
112             database=data_source.redshift_options.database,
113         )
114 
115     # Note: Python requires redefining hash in child classes that override __eq__
116     def __hash__(self):
117         return super().__hash__()
118 
119     def __eq__(self, other):
120         if not isinstance(other, RedshiftSource):
121             raise TypeError(
122                 "Comparisons should only involve RedshiftSource class objects."
123             )
124 
125         return (
126             super().__eq__(other)
127             and self.redshift_options.table == other.redshift_options.table
128             and self.redshift_options.schema == other.redshift_options.schema
129             and self.redshift_options.query == other.redshift_options.query
130             and self.redshift_options.database == other.redshift_options.database
131         )
132 
133     @property
134     def table(self):
135         """Returns the table of this Redshift source."""
136         return self.redshift_options.table
137 
138     @property
139     def schema(self):
140         """Returns the schema of this Redshift source."""
141         return self.redshift_options.schema
142 
143     @property
144     def query(self):
145         """Returns the Redshift query of this Redshift source."""
146         return self.redshift_options.query
147 
148     @property
149     def database(self):
150         """Returns the Redshift database of this Redshift source."""
151         return self.redshift_options.database
152 
153     def to_proto(self) -> DataSourceProto:
154         """
155         Converts a RedshiftSource object to its protobuf representation.
156 
157         Returns:
158             A DataSourceProto object.
159         """
160         data_source_proto = DataSourceProto(
161             name=self.name,
162             type=DataSourceProto.BATCH_REDSHIFT,
163             field_mapping=self.field_mapping,
164             redshift_options=self.redshift_options.to_proto(),
165             description=self.description,
166             tags=self.tags,
167             owner=self.owner,
168             timestamp_field=self.timestamp_field,
169             created_timestamp_column=self.created_timestamp_column,
170         )
171 
172         return data_source_proto
173 
174     def validate(self, config: RepoConfig):
175         # As long as the query gets successfully executed, or the table exists,
176         # the data source is validated. We don't need the results though.
177         self.get_table_column_names_and_types(config)
178 
179     def get_table_query_string(self) -> str:
180         """Returns a string that can directly be used to reference this table in SQL."""
181         if self.table:
182             return f'"{self.schema}"."{self.table}"'
183         else:
184             return f"({self.query})"
185 
186     @staticmethod
187     def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:
188         return type_map.redshift_to_feast_value_type
189 
190     def get_table_column_names_and_types(
191         self, config: RepoConfig
192     ) -> Iterable[Tuple[str, str]]:
193         """
194         Returns a mapping of column names to types for this Redshift source.
195 
196         Args:
197             config: A RepoConfig describing the feature repo
198         """
199         from botocore.exceptions import ClientError
200 
201         from feast.infra.offline_stores.redshift import RedshiftOfflineStoreConfig
202         from feast.infra.utils import aws_utils
203 
204         assert isinstance(config.offline_store, RedshiftOfflineStoreConfig)
205 
206         client = aws_utils.get_redshift_data_client(config.offline_store.region)
207         if self.table:
208             try:
209                 table = client.describe_table(
210                     ClusterIdentifier=config.offline_store.cluster_id,
211                     Database=(
212                         self.database
213                         if self.database
214                         else config.offline_store.database
215                     ),
216                     DbUser=config.offline_store.user,
217                     Table=self.table,
218                     Schema=self.schema,
219                 )
220             except ClientError as e:
221                 if e.response["Error"]["Code"] == "ValidationException":
222                     raise RedshiftCredentialsError() from e
223                 raise
224 
225             # The API returns valid JSON with empty column list when the table doesn't exist
226             if len(table["ColumnList"]) == 0:
227                 raise DataSourceNotFoundException(self.table)
228 
229             columns = table["ColumnList"]
230         else:
231             statement_id = aws_utils.execute_redshift_statement(
232                 client,
233                 config.offline_store.cluster_id,
234                 self.database if self.database else config.offline_store.database,
235                 config.offline_store.user,
236                 f"SELECT * FROM ({self.query}) LIMIT 1",
237             )
238             columns = aws_utils.get_redshift_statement_result(client, statement_id)[
239                 "ColumnMetadata"
240             ]
241 
242         return [(column["name"], column["typeName"].upper()) for column in columns]
243 
244 
245 class RedshiftOptions:
246     """
247     Configuration options for a Redshift data source.
248     """
249 
250     def __init__(
251         self,
252         table: Optional[str],
253         schema: Optional[str],
254         query: Optional[str],
255         database: Optional[str],
256     ):
257         self.table = table or ""
258         self.schema = schema or ""
259         self.query = query or ""
260         self.database = database or ""
261 
262     @classmethod
263     def from_proto(cls, redshift_options_proto: DataSourceProto.RedshiftOptions):
264         """
265         Creates a RedshiftOptions from a protobuf representation of a Redshift option.
266 
267         Args:
268             redshift_options_proto: A protobuf representation of a DataSource
269 
270         Returns:
271             A RedshiftOptions object based on the redshift_options protobuf.
272         """
273         redshift_options = cls(
274             table=redshift_options_proto.table,
275             schema=redshift_options_proto.schema,
276             query=redshift_options_proto.query,
277             database=redshift_options_proto.database,
278         )
279 
280         return redshift_options
281 
282     def to_proto(self) -> DataSourceProto.RedshiftOptions:
283         """
284         Converts an RedshiftOptionsProto object to its protobuf representation.
285 
286         Returns:
287             A RedshiftOptionsProto protobuf.
288         """
289         redshift_options_proto = DataSourceProto.RedshiftOptions(
290             table=self.table,
291             schema=self.schema,
292             query=self.query,
293             database=self.database,
294         )
295 
296         return redshift_options_proto
297 
298 
299 class SavedDatasetRedshiftStorage(SavedDatasetStorage):
300     _proto_attr_name = "redshift_storage"
301 
302     redshift_options: RedshiftOptions
303 
304     def __init__(self, table_ref: str):
305         self.redshift_options = RedshiftOptions(
306             table=table_ref, schema=None, query=None, database=None
307         )
308 
309     @staticmethod
310     def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:
311 
312         return SavedDatasetRedshiftStorage(
313             table_ref=RedshiftOptions.from_proto(storage_proto.redshift_storage).table
314         )
315 
316     def to_proto(self) -> SavedDatasetStorageProto:
317         return SavedDatasetStorageProto(
318             redshift_storage=self.redshift_options.to_proto()
319         )
320 
321     def to_data_source(self) -> DataSource:
322         return RedshiftSource(table=self.redshift_options.table)
323 
324 
325 class RedshiftLoggingDestination(LoggingDestination):
326     _proto_kind = "redshift_destination"
327 
328     table_name: str
329 
330     def __init__(self, *, table_name: str):
331         self.table_name = table_name
332 
333     @classmethod
334     def from_proto(cls, config_proto: LoggingConfigProto) -> "LoggingDestination":
335         return RedshiftLoggingDestination(
336             table_name=config_proto.redshift_destination.table_name,
337         )
338 
339     def to_proto(self) -> LoggingConfigProto:
340         return LoggingConfigProto(
341             redshift_destination=LoggingConfigProto.RedshiftDestination(
342                 table_name=self.table_name
343             )
344         )
345 
346     def to_data_source(self) -> DataSource:
347         return RedshiftSource(table=self.table_name)
348
```
--- 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/offline_stores/redshift_source.py b/sdk/python/feast/infra/offline_stores/redshift_source.py
--- a/sdk/python/feast/infra/offline_stores/redshift_source.py
+++ b/sdk/python/feast/infra/offline_stores/redshift_source.py
@@ -206,7 +206,8 @@
         client = aws_utils.get_redshift_data_client(config.offline_store.region)
         if self.table:
             try:
-                table = client.describe_table(
+                paginator = client.get_paginator("describe_table")
+                response_iterator = paginator.paginate(
                     ClusterIdentifier=config.offline_store.cluster_id,
                     Database=(
                         self.database
@@ -217,6 +218,7 @@
                     Table=self.table,
                     Schema=self.schema,
                 )
+                table = response_iterator.build_full_result()
             except ClientError as e:
                 if e.response["Error"]["Code"] == "ValidationException":
                     raise RedshiftCredentialsError() from e

{"golden_diff": "diff --git a/sdk/python/feast/infra/offline_stores/redshift_source.py b/sdk/python/feast/infra/offline_stores/redshift_source.py\n--- a/sdk/python/feast/infra/offline_stores/redshift_source.py\n+++ b/sdk/python/feast/infra/offline_stores/redshift_source.py\n@@ -206,7 +206,8 @@\n         client = aws_utils.get_redshift_data_client(config.offline_store.region)\n         if self.table:\n             try:\n-                table = client.describe_table(\n+                paginator = client.get_paginator(\"describe_table\")\n+                response_iterator = paginator.paginate(\n                     ClusterIdentifier=config.offline_store.cluster_id,\n                     Database=(\n                         self.database\n@@ -217,6 +218,7 @@\n                     Table=self.table,\n                     Schema=self.schema,\n                 )\n+                table = response_iterator.build_full_result()\n             except ClientError as e:\n                 if e.response[\"Error\"][\"Code\"] == \"ValidationException\":\n                     raise RedshiftCredentialsError() from e\n", "issue": "Feast push (Redshift/DynamoDb) not work with PushMode.ONLINE_AND_OFFLINE when more than 500 columns\n## Expected Behavior \r\nCurrently, we have a push source with Redshift Offline Store and DynamoDb  Online Store. \r\nWe built our view with more than 500 columns. Around 750 columns. \r\n\r\nWe expected to ingest data in dynamo and redshift when we run \r\n`fs.push(\"push_source\", df, to=PushMode.ONLINE_AND_OFFLINE)`\r\n\r\n## Current Behavior\r\n\r\nPush command raise an issue like` [ERROR] ValueError: The input dataframe has columns ..`\r\nThis issue come from `get_table_column_names_and_types `method in `write_to_offline_store` method.\r\nIn the method, we check if `if set(input_columns) != set(source_columns)` and raise the below issue if there are diff. \r\n\r\nIn case with more than 500 columns we get a diff because source_columns come from `get_table_column_names_and_types` method result where the result is define by MaxResults parameters. \r\n\r\n## Steps to reproduce\r\n```\r\nentity= Entity(\r\n    name=\"entity\",\r\n    join_keys=[\"entity_id\"],\r\n    value_type=ValueType.INT64,\r\n)\r\n\r\npush_source = PushSource(\r\n    name=\"push_source\",\r\n    batch_source=RedshiftSource(\r\n        table=\"fs_push_view\",\r\n        timestamp_field=\"datecreation\",\r\n        created_timestamp_column=\"created_at\"),\r\n)\r\n\r\nbesoin_embedding_push_view = FeatureView(\r\n    name=\"push_view\",\r\n    entities=[entity],\r\n    schema=[Field(name=f\"field_{dim}\", dtype=types.Float64) for dim in range(768)],\r\n    source=push_source \r\n)\r\n\r\nfs.push(\"push_source\", df, to=PushMode.ONLINE_AND_OFFLINE)\r\n\r\n```\r\n\r\n### Specifications\r\n\r\n- Version: 0.25.0\r\n- Platform: AWS\r\n- Subsystem:\r\n\r\n## Possible Solution\r\n\r\nIn my mind, we have two solutions:\r\n\r\n- Set higher MaxResults in describe_table method\r\n- Use NextToken to iterate through results\r\n\r\n\n", "before_files": [{"content": "from typing import Callable, Dict, Iterable, Optional, Tuple\n\nfrom typeguard import typechecked\n\nfrom feast import type_map\nfrom feast.data_source import DataSource\nfrom feast.errors import (\n    DataSourceNoNameException,\n    DataSourceNotFoundException,\n    RedshiftCredentialsError,\n)\nfrom feast.feature_logging import LoggingDestination\nfrom feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto\nfrom feast.protos.feast.core.FeatureService_pb2 import (\n    LoggingConfig as LoggingConfigProto,\n)\nfrom feast.protos.feast.core.SavedDataset_pb2 import (\n    SavedDatasetStorage as SavedDatasetStorageProto,\n)\nfrom feast.repo_config import RepoConfig\nfrom feast.saved_dataset import SavedDatasetStorage\nfrom feast.value_type import ValueType\n\n\n@typechecked\nclass RedshiftSource(DataSource):\n    def __init__(\n        self,\n        *,\n        name: Optional[str] = None,\n        timestamp_field: Optional[str] = \"\",\n        table: Optional[str] = None,\n        schema: Optional[str] = None,\n        created_timestamp_column: Optional[str] = \"\",\n        field_mapping: Optional[Dict[str, str]] = None,\n        query: Optional[str] = None,\n        description: Optional[str] = \"\",\n        tags: Optional[Dict[str, str]] = None,\n        owner: Optional[str] = \"\",\n        database: Optional[str] = \"\",\n    ):\n        \"\"\"\n        Creates a RedshiftSource object.\n\n        Args:\n            name (optional): Name for the source. Defaults to the table if not specified, in which\n                case the table must be specified.\n            timestamp_field (optional): Event timestamp field used for point in time\n                joins of feature values.\n            table (optional): Redshift table where the features are stored. Exactly one of 'table'\n                and 'query' must be specified.\n            schema (optional): Redshift schema in which the table is located.\n            created_timestamp_column (optional): Timestamp column indicating when the\n                row was created, used for deduplicating rows.\n            field_mapping (optional): A dictionary mapping of column names in this data\n                source to column names in a feature table or view.\n            query (optional): The query to be executed to obtain the features. Exactly one of 'table'\n                and 'query' must be specified.\n            description (optional): A human-readable description.\n            tags (optional): A dictionary of key-value pairs to store arbitrary metadata.\n            owner (optional): The owner of the redshift source, typically the email of the primary\n                maintainer.\n            database (optional): The Redshift database name.\n        \"\"\"\n        if table is None and query is None:\n            raise ValueError('No \"table\" or \"query\" argument provided.')\n\n        # The default Redshift schema is named \"public\".\n        _schema = \"public\" if table and not schema else schema\n        self.redshift_options = RedshiftOptions(\n            table=table, schema=_schema, query=query, database=database\n        )\n\n        # If no name, use the table as the default name.\n        if name is None and table is None:\n            raise DataSourceNoNameException()\n        name = name or table\n        assert name\n\n        super().__init__(\n            name=name,\n            timestamp_field=timestamp_field,\n            created_timestamp_column=created_timestamp_column,\n            field_mapping=field_mapping,\n            description=description,\n            tags=tags,\n            owner=owner,\n        )\n\n    @staticmethod\n    def from_proto(data_source: DataSourceProto):\n        \"\"\"\n        Creates a RedshiftSource from a protobuf representation of a RedshiftSource.\n\n        Args:\n            data_source: A protobuf representation of a RedshiftSource\n\n        Returns:\n            A RedshiftSource object based on the data_source protobuf.\n        \"\"\"\n        return RedshiftSource(\n            name=data_source.name,\n            timestamp_field=data_source.timestamp_field,\n            table=data_source.redshift_options.table,\n            schema=data_source.redshift_options.schema,\n            created_timestamp_column=data_source.created_timestamp_column,\n            field_mapping=dict(data_source.field_mapping),\n            query=data_source.redshift_options.query,\n            description=data_source.description,\n            tags=dict(data_source.tags),\n            owner=data_source.owner,\n            database=data_source.redshift_options.database,\n        )\n\n    # Note: Python requires redefining hash in child classes that override __eq__\n    def __hash__(self):\n        return super().__hash__()\n\n    def __eq__(self, other):\n        if not isinstance(other, RedshiftSource):\n            raise TypeError(\n                \"Comparisons should only involve RedshiftSource class objects.\"\n            )\n\n        return (\n            super().__eq__(other)\n            and self.redshift_options.table == other.redshift_options.table\n            and self.redshift_options.schema == other.redshift_options.schema\n            and self.redshift_options.query == other.redshift_options.query\n            and self.redshift_options.database == other.redshift_options.database\n        )\n\n    @property\n    def table(self):\n        \"\"\"Returns the table of this Redshift source.\"\"\"\n        return self.redshift_options.table\n\n    @property\n    def schema(self):\n        \"\"\"Returns the schema of this Redshift source.\"\"\"\n        return self.redshift_options.schema\n\n    @property\n    def query(self):\n        \"\"\"Returns the Redshift query of this Redshift source.\"\"\"\n        return self.redshift_options.query\n\n    @property\n    def database(self):\n        \"\"\"Returns the Redshift database of this Redshift source.\"\"\"\n        return self.redshift_options.database\n\n    def to_proto(self) -> DataSourceProto:\n        \"\"\"\n        Converts a RedshiftSource object to its protobuf representation.\n\n        Returns:\n            A DataSourceProto object.\n        \"\"\"\n        data_source_proto = DataSourceProto(\n            name=self.name,\n            type=DataSourceProto.BATCH_REDSHIFT,\n            field_mapping=self.field_mapping,\n            redshift_options=self.redshift_options.to_proto(),\n            description=self.description,\n            tags=self.tags,\n            owner=self.owner,\n            timestamp_field=self.timestamp_field,\n            created_timestamp_column=self.created_timestamp_column,\n        )\n\n        return data_source_proto\n\n    def validate(self, config: RepoConfig):\n        # As long as the query gets successfully executed, or the table exists,\n        # the data source is validated. We don't need the results though.\n        self.get_table_column_names_and_types(config)\n\n    def get_table_query_string(self) -> str:\n        \"\"\"Returns a string that can directly be used to reference this table in SQL.\"\"\"\n        if self.table:\n            return f'\"{self.schema}\".\"{self.table}\"'\n        else:\n            return f\"({self.query})\"\n\n    @staticmethod\n    def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:\n        return type_map.redshift_to_feast_value_type\n\n    def get_table_column_names_and_types(\n        self, config: RepoConfig\n    ) -> Iterable[Tuple[str, str]]:\n        \"\"\"\n        Returns a mapping of column names to types for this Redshift source.\n\n        Args:\n            config: A RepoConfig describing the feature repo\n        \"\"\"\n        from botocore.exceptions import ClientError\n\n        from feast.infra.offline_stores.redshift import RedshiftOfflineStoreConfig\n        from feast.infra.utils import aws_utils\n\n        assert isinstance(config.offline_store, RedshiftOfflineStoreConfig)\n\n        client = aws_utils.get_redshift_data_client(config.offline_store.region)\n        if self.table:\n            try:\n                table = client.describe_table(\n                    ClusterIdentifier=config.offline_store.cluster_id,\n                    Database=(\n                        self.database\n                        if self.database\n                        else config.offline_store.database\n                    ),\n                    DbUser=config.offline_store.user,\n                    Table=self.table,\n                    Schema=self.schema,\n                )\n            except ClientError as e:\n                if e.response[\"Error\"][\"Code\"] == \"ValidationException\":\n                    raise RedshiftCredentialsError() from e\n                raise\n\n            # The API returns valid JSON with empty column list when the table doesn't exist\n            if len(table[\"ColumnList\"]) == 0:\n                raise DataSourceNotFoundException(self.table)\n\n            columns = table[\"ColumnList\"]\n        else:\n            statement_id = aws_utils.execute_redshift_statement(\n                client,\n                config.offline_store.cluster_id,\n                self.database if self.database else config.offline_store.database,\n                config.offline_store.user,\n                f\"SELECT * FROM ({self.query}) LIMIT 1\",\n            )\n            columns = aws_utils.get_redshift_statement_result(client, statement_id)[\n                \"ColumnMetadata\"\n            ]\n\n        return [(column[\"name\"], column[\"typeName\"].upper()) for column in columns]\n\n\nclass RedshiftOptions:\n    \"\"\"\n    Configuration options for a Redshift data source.\n    \"\"\"\n\n    def __init__(\n        self,\n        table: Optional[str],\n        schema: Optional[str],\n        query: Optional[str],\n        database: Optional[str],\n    ):\n        self.table = table or \"\"\n        self.schema = schema or \"\"\n        self.query = query or \"\"\n        self.database = database or \"\"\n\n    @classmethod\n    def from_proto(cls, redshift_options_proto: DataSourceProto.RedshiftOptions):\n        \"\"\"\n        Creates a RedshiftOptions from a protobuf representation of a Redshift option.\n\n        Args:\n            redshift_options_proto: A protobuf representation of a DataSource\n\n        Returns:\n            A RedshiftOptions object based on the redshift_options protobuf.\n        \"\"\"\n        redshift_options = cls(\n            table=redshift_options_proto.table,\n            schema=redshift_options_proto.schema,\n            query=redshift_options_proto.query,\n            database=redshift_options_proto.database,\n        )\n\n        return redshift_options\n\n    def to_proto(self) -> DataSourceProto.RedshiftOptions:\n        \"\"\"\n        Converts an RedshiftOptionsProto object to its protobuf representation.\n\n        Returns:\n            A RedshiftOptionsProto protobuf.\n        \"\"\"\n        redshift_options_proto = DataSourceProto.RedshiftOptions(\n            table=self.table,\n            schema=self.schema,\n            query=self.query,\n            database=self.database,\n        )\n\n        return redshift_options_proto\n\n\nclass SavedDatasetRedshiftStorage(SavedDatasetStorage):\n    _proto_attr_name = \"redshift_storage\"\n\n    redshift_options: RedshiftOptions\n\n    def __init__(self, table_ref: str):\n        self.redshift_options = RedshiftOptions(\n            table=table_ref, schema=None, query=None, database=None\n        )\n\n    @staticmethod\n    def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:\n\n        return SavedDatasetRedshiftStorage(\n            table_ref=RedshiftOptions.from_proto(storage_proto.redshift_storage).table\n        )\n\n    def to_proto(self) -> SavedDatasetStorageProto:\n        return SavedDatasetStorageProto(\n            redshift_storage=self.redshift_options.to_proto()\n        )\n\n    def to_data_source(self) -> DataSource:\n        return RedshiftSource(table=self.redshift_options.table)\n\n\nclass RedshiftLoggingDestination(LoggingDestination):\n    _proto_kind = \"redshift_destination\"\n\n    table_name: str\n\n    def __init__(self, *, table_name: str):\n        self.table_name = table_name\n\n    @classmethod\n    def from_proto(cls, config_proto: LoggingConfigProto) -> \"LoggingDestination\":\n        return RedshiftLoggingDestination(\n            table_name=config_proto.redshift_destination.table_name,\n        )\n\n    def to_proto(self) -> LoggingConfigProto:\n        return LoggingConfigProto(\n            redshift_destination=LoggingConfigProto.RedshiftDestination(\n                table_name=self.table_name\n            )\n        )\n\n    def to_data_source(self) -> DataSource:\n        return RedshiftSource(table=self.table_name)\n", "path": "sdk/python/feast/infra/offline_stores/redshift_source.py"}], "after_files": [{"content": "from typing import Callable, Dict, Iterable, Optional, Tuple\n\nfrom typeguard import typechecked\n\nfrom feast import type_map\nfrom feast.data_source import DataSource\nfrom feast.errors import (\n    DataSourceNoNameException,\n    DataSourceNotFoundException,\n    RedshiftCredentialsError,\n)\nfrom feast.feature_logging import LoggingDestination\nfrom feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto\nfrom feast.protos.feast.core.FeatureService_pb2 import (\n    LoggingConfig as LoggingConfigProto,\n)\nfrom feast.protos.feast.core.SavedDataset_pb2 import (\n    SavedDatasetStorage as SavedDatasetStorageProto,\n)\nfrom feast.repo_config import RepoConfig\nfrom feast.saved_dataset import SavedDatasetStorage\nfrom feast.value_type import ValueType\n\n\n@typechecked\nclass RedshiftSource(DataSource):\n    def __init__(\n        self,\n        *,\n        name: Optional[str] = None,\n        timestamp_field: Optional[str] = \"\",\n        table: Optional[str] = None,\n        schema: Optional[str] = None,\n        created_timestamp_column: Optional[str] = \"\",\n        field_mapping: Optional[Dict[str, str]] = None,\n        query: Optional[str] = None,\n        description: Optional[str] = \"\",\n        tags: Optional[Dict[str, str]] = None,\n        owner: Optional[str] = \"\",\n        database: Optional[str] = \"\",\n    ):\n        \"\"\"\n        Creates a RedshiftSource object.\n\n        Args:\n            name (optional): Name for the source. Defaults to the table if not specified, in which\n                case the table must be specified.\n            timestamp_field (optional): Event timestamp field used for point in time\n                joins of feature values.\n            table (optional): Redshift table where the features are stored. Exactly one of 'table'\n                and 'query' must be specified.\n            schema (optional): Redshift schema in which the table is located.\n            created_timestamp_column (optional): Timestamp column indicating when the\n                row was created, used for deduplicating rows.\n            field_mapping (optional): A dictionary mapping of column names in this data\n                source to column names in a feature table or view.\n            query (optional): The query to be executed to obtain the features. Exactly one of 'table'\n                and 'query' must be specified.\n            description (optional): A human-readable description.\n            tags (optional): A dictionary of key-value pairs to store arbitrary metadata.\n            owner (optional): The owner of the redshift source, typically the email of the primary\n                maintainer.\n            database (optional): The Redshift database name.\n        \"\"\"\n        if table is None and query is None:\n            raise ValueError('No \"table\" or \"query\" argument provided.')\n\n        # The default Redshift schema is named \"public\".\n        _schema = \"public\" if table and not schema else schema\n        self.redshift_options = RedshiftOptions(\n            table=table, schema=_schema, query=query, database=database\n        )\n\n        # If no name, use the table as the default name.\n        if name is None and table is None:\n            raise DataSourceNoNameException()\n        name = name or table\n        assert name\n\n        super().__init__(\n            name=name,\n            timestamp_field=timestamp_field,\n            created_timestamp_column=created_timestamp_column,\n            field_mapping=field_mapping,\n            description=description,\n            tags=tags,\n            owner=owner,\n        )\n\n    @staticmethod\n    def from_proto(data_source: DataSourceProto):\n        \"\"\"\n        Creates a RedshiftSource from a protobuf representation of a RedshiftSource.\n\n        Args:\n            data_source: A protobuf representation of a RedshiftSource\n\n        Returns:\n            A RedshiftSource object based on the data_source protobuf.\n        \"\"\"\n        return RedshiftSource(\n            name=data_source.name,\n            timestamp_field=data_source.timestamp_field,\n            table=data_source.redshift_options.table,\n            schema=data_source.redshift_options.schema,\n            created_timestamp_column=data_source.created_timestamp_column,\n            field_mapping=dict(data_source.field_mapping),\n            query=data_source.redshift_options.query,\n            description=data_source.description,\n            tags=dict(data_source.tags),\n            owner=data_source.owner,\n            database=data_source.redshift_options.database,\n        )\n\n    # Note: Python requires redefining hash in child classes that override __eq__\n    def __hash__(self):\n        return super().__hash__()\n\n    def __eq__(self, other):\n        if not isinstance(other, RedshiftSource):\n            raise TypeError(\n                \"Comparisons should only involve RedshiftSource class objects.\"\n            )\n\n        return (\n            super().__eq__(other)\n            and self.redshift_options.table == other.redshift_options.table\n            and self.redshift_options.schema == other.redshift_options.schema\n            and self.redshift_options.query == other.redshift_options.query\n            and self.redshift_options.database == other.redshift_options.database\n        )\n\n    @property\n    def table(self):\n        \"\"\"Returns the table of this Redshift source.\"\"\"\n        return self.redshift_options.table\n\n    @property\n    def schema(self):\n        \"\"\"Returns the schema of this Redshift source.\"\"\"\n        return self.redshift_options.schema\n\n    @property\n    def query(self):\n        \"\"\"Returns the Redshift query of this Redshift source.\"\"\"\n        return self.redshift_options.query\n\n    @property\n    def database(self):\n        \"\"\"Returns the Redshift database of this Redshift source.\"\"\"\n        return self.redshift_options.database\n\n    def to_proto(self) -> DataSourceProto:\n        \"\"\"\n        Converts a RedshiftSource object to its protobuf representation.\n\n        Returns:\n            A DataSourceProto object.\n        \"\"\"\n        data_source_proto = DataSourceProto(\n            name=self.name,\n            type=DataSourceProto.BATCH_REDSHIFT,\n            field_mapping=self.field_mapping,\n            redshift_options=self.redshift_options.to_proto(),\n            description=self.description,\n            tags=self.tags,\n            owner=self.owner,\n            timestamp_field=self.timestamp_field,\n            created_timestamp_column=self.created_timestamp_column,\n        )\n\n        return data_source_proto\n\n    def validate(self, config: RepoConfig):\n        # As long as the query gets successfully executed, or the table exists,\n        # the data source is validated. We don't need the results though.\n        self.get_table_column_names_and_types(config)\n\n    def get_table_query_string(self) -> str:\n        \"\"\"Returns a string that can directly be used to reference this table in SQL.\"\"\"\n        if self.table:\n            return f'\"{self.schema}\".\"{self.table}\"'\n        else:\n            return f\"({self.query})\"\n\n    @staticmethod\n    def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:\n        return type_map.redshift_to_feast_value_type\n\n    def get_table_column_names_and_types(\n        self, config: RepoConfig\n    ) -> Iterable[Tuple[str, str]]:\n        \"\"\"\n        Returns a mapping of column names to types for this Redshift source.\n\n        Args:\n            config: A RepoConfig describing the feature repo\n        \"\"\"\n        from botocore.exceptions import ClientError\n\n        from feast.infra.offline_stores.redshift import RedshiftOfflineStoreConfig\n        from feast.infra.utils import aws_utils\n\n        assert isinstance(config.offline_store, RedshiftOfflineStoreConfig)\n\n        client = aws_utils.get_redshift_data_client(config.offline_store.region)\n        if self.table:\n            try:\n                paginator = client.get_paginator(\"describe_table\")\n                response_iterator = paginator.paginate(\n                    ClusterIdentifier=config.offline_store.cluster_id,\n                    Database=(\n                        self.database\n                        if self.database\n                        else config.offline_store.database\n                    ),\n                    DbUser=config.offline_store.user,\n                    Table=self.table,\n                    Schema=self.schema,\n                )\n                table = response_iterator.build_full_result()\n            except ClientError as e:\n                if e.response[\"Error\"][\"Code\"] == \"ValidationException\":\n                    raise RedshiftCredentialsError() from e\n                raise\n\n            # The API returns valid JSON with empty column list when the table doesn't exist\n            if len(table[\"ColumnList\"]) == 0:\n                raise DataSourceNotFoundException(self.table)\n\n            columns = table[\"ColumnList\"]\n        else:\n            statement_id = aws_utils.execute_redshift_statement(\n                client,\n                config.offline_store.cluster_id,\n                self.database if self.database else config.offline_store.database,\n                config.offline_store.user,\n                f\"SELECT * FROM ({self.query}) LIMIT 1\",\n            )\n            columns = aws_utils.get_redshift_statement_result(client, statement_id)[\n                \"ColumnMetadata\"\n            ]\n\n        return [(column[\"name\"], column[\"typeName\"].upper()) for column in columns]\n\n\nclass RedshiftOptions:\n    \"\"\"\n    Configuration options for a Redshift data source.\n    \"\"\"\n\n    def __init__(\n        self,\n        table: Optional[str],\n        schema: Optional[str],\n        query: Optional[str],\n        database: Optional[str],\n    ):\n        self.table = table or \"\"\n        self.schema = schema or \"\"\n        self.query = query or \"\"\n        self.database = database or \"\"\n\n    @classmethod\n    def from_proto(cls, redshift_options_proto: DataSourceProto.RedshiftOptions):\n        \"\"\"\n        Creates a RedshiftOptions from a protobuf representation of a Redshift option.\n\n        Args:\n            redshift_options_proto: A protobuf representation of a DataSource\n\n        Returns:\n            A RedshiftOptions object based on the redshift_options protobuf.\n        \"\"\"\n        redshift_options = cls(\n            table=redshift_options_proto.table,\n            schema=redshift_options_proto.schema,\n            query=redshift_options_proto.query,\n            database=redshift_options_proto.database,\n        )\n\n        return redshift_options\n\n    def to_proto(self) -> DataSourceProto.RedshiftOptions:\n        \"\"\"\n        Converts an RedshiftOptionsProto object to its protobuf representation.\n\n        Returns:\n            A RedshiftOptionsProto protobuf.\n        \"\"\"\n        redshift_options_proto = DataSourceProto.RedshiftOptions(\n            table=self.table,\n            schema=self.schema,\n            query=self.query,\n            database=self.database,\n        )\n\n        return redshift_options_proto\n\n\nclass SavedDatasetRedshiftStorage(SavedDatasetStorage):\n    _proto_attr_name = \"redshift_storage\"\n\n    redshift_options: RedshiftOptions\n\n    def __init__(self, table_ref: str):\n        self.redshift_options = RedshiftOptions(\n            table=table_ref, schema=None, query=None, database=None\n        )\n\n    @staticmethod\n    def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:\n\n        return SavedDatasetRedshiftStorage(\n            table_ref=RedshiftOptions.from_proto(storage_proto.redshift_storage).table\n        )\n\n    def to_proto(self) -> SavedDatasetStorageProto:\n        return SavedDatasetStorageProto(\n            redshift_storage=self.redshift_options.to_proto()\n        )\n\n    def to_data_source(self) -> DataSource:\n        return RedshiftSource(table=self.redshift_options.table)\n\n\nclass RedshiftLoggingDestination(LoggingDestination):\n    _proto_kind = \"redshift_destination\"\n\n    table_name: str\n\n    def __init__(self, *, table_name: str):\n        self.table_name = table_name\n\n    @classmethod\n    def from_proto(cls, config_proto: LoggingConfigProto) -> \"LoggingDestination\":\n        return RedshiftLoggingDestination(\n            table_name=config_proto.redshift_destination.table_name,\n        )\n\n    def to_proto(self) -> LoggingConfigProto:\n        return LoggingConfigProto(\n            redshift_destination=LoggingConfigProto.RedshiftDestination(\n                table_name=self.table_name\n            )\n        )\n\n    def to_data_source(self) -> DataSource:\n        return RedshiftSource(table=self.table_name)\n", "path": "sdk/python/feast/infra/offline_stores/redshift_source.py"}]}