We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Parquet Schema Inference only supports File, not directory
When using a FileSource that is in Parquet format, if the source happens to be a directory of partitioned Parquet files, the following lines throw an error:

https://github.com/feast-dev/feast/blob/01d3568168bb9febb9fbda4988283b3886c32a31/sdk/python/feast/infra/offline_stores/file_source.py#L182-L184

`OSError: Expected file path, but /home/ubuntu/project/data/driver_stats_partitioned is a directory`

How to replicate:

1. Start with a demo feast project (`feast init`)
2. Create a partitioned Parquet Dataset.  Use the following to create a dataset with only a single timestamp for inference
```
import pyarrow.parquet as pq
df = pq.read_table("./data/driver_stats.parquet")
df = df.drop(["created"])
pq.write_to_dataset(df, "./data/driver_stats_partitioned")
```
3. Update the file source in `example.py` to look like this:
```
driver_hourly_stats = FileSource(
    path="/home/ubuntu/cado-feast/feature_store/exciting_sunbeam/data/driver_stats_partitioned2",
)
```

4. Run `feast apply`
For now, I've been able to fix by updating the above lines to:
```
schema = ParquetDataset(
    path if filesystem is None else filesystem.open_input_file(path)
).schema.to_arrow_schema()
```

--- 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/file_source.py`
Content:
```
1 import warnings
2 from typing import Callable, Dict, Iterable, List, Optional, Tuple
3 
4 from pyarrow._fs import FileSystem
5 from pyarrow._s3fs import S3FileSystem
6 from pyarrow.parquet import ParquetFile
7 
8 from feast import type_map
9 from feast.data_format import FileFormat, ParquetFormat
10 from feast.data_source import DataSource
11 from feast.feature_logging import LoggingDestination
12 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
13 from feast.protos.feast.core.FeatureService_pb2 import (
14     LoggingConfig as LoggingConfigProto,
15 )
16 from feast.protos.feast.core.SavedDataset_pb2 import (
17     SavedDatasetStorage as SavedDatasetStorageProto,
18 )
19 from feast.repo_config import RepoConfig
20 from feast.saved_dataset import SavedDatasetStorage
21 from feast.value_type import ValueType
22 
23 
24 class FileSource(DataSource):
25     def __init__(
26         self,
27         *args,
28         path: Optional[str] = None,
29         event_timestamp_column: Optional[str] = "",
30         file_format: Optional[FileFormat] = None,
31         created_timestamp_column: Optional[str] = "",
32         field_mapping: Optional[Dict[str, str]] = None,
33         date_partition_column: Optional[str] = "",
34         s3_endpoint_override: Optional[str] = None,
35         name: Optional[str] = "",
36         description: Optional[str] = "",
37         tags: Optional[Dict[str, str]] = None,
38         owner: Optional[str] = "",
39         timestamp_field: Optional[str] = "",
40     ):
41         """Create a FileSource from a file containing feature data. Only Parquet format supported.
42 
43         Args:
44 
45             path: File path to file containing feature data. Must contain an event_timestamp column, entity columns and
46                 feature columns.
47             event_timestamp_column(optional): (Deprecated) Event timestamp column used for point in time joins of feature values.
48             created_timestamp_column (optional): Timestamp column when row was created, used for deduplicating rows.
49             file_format (optional): Explicitly set the file format. Allows Feast to bypass inferring the file format.
50             field_mapping: A dictionary mapping of column names in this data source to feature names in a feature table
51                 or view. Only used for feature columns, not entities or timestamp columns.
52             date_partition_column (optional): Timestamp column used for partitioning.
53             s3_endpoint_override (optional): Overrides AWS S3 enpoint with custom S3 storage
54             name (optional): Name for the file source. Defaults to the path.
55             description (optional): A human-readable description.
56             tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
57             owner (optional): The owner of the file source, typically the email of the primary
58                 maintainer.
59             timestamp_field (optional): Event timestamp foe;d used for point in time
60                 joins of feature values.
61 
62         Examples:
63             >>> from feast import FileSource
64             >>> file_source = FileSource(path="my_features.parquet", timestamp_field="event_timestamp")
65         """
66         positional_attributes = ["path"]
67         _path = path
68         if args:
69             if args:
70                 warnings.warn(
71                     (
72                         "File Source parameters should be specified as a keyword argument instead of a positional arg."
73                         "Feast 0.23+ will not support positional arguments to construct File sources"
74                     ),
75                     DeprecationWarning,
76                 )
77                 if len(args) > len(positional_attributes):
78                     raise ValueError(
79                         f"Only {', '.join(positional_attributes)} are allowed as positional args when defining "
80                         f"File sources, for backwards compatibility."
81                     )
82                 if len(args) >= 1:
83                     _path = args[0]
84         if _path is None:
85             raise ValueError(
86                 'No "path" argument provided. Please set "path" to the location of your file source.'
87             )
88         self.file_options = FileOptions(
89             file_format=file_format,
90             uri=_path,
91             s3_endpoint_override=s3_endpoint_override,
92         )
93 
94         if date_partition_column:
95             warnings.warn(
96                 (
97                     "The argument 'date_partition_column' is not supported for File sources."
98                     "It will be removed in Feast 0.23+"
99                 ),
100                 DeprecationWarning,
101             )
102 
103         super().__init__(
104             name=name if name else path,
105             event_timestamp_column=event_timestamp_column,
106             created_timestamp_column=created_timestamp_column,
107             field_mapping=field_mapping,
108             description=description,
109             tags=tags,
110             owner=owner,
111             timestamp_field=timestamp_field,
112         )
113 
114     # Note: Python requires redefining hash in child classes that override __eq__
115     def __hash__(self):
116         return super().__hash__()
117 
118     def __eq__(self, other):
119         if not isinstance(other, FileSource):
120             raise TypeError("Comparisons should only involve FileSource class objects.")
121 
122         return (
123             super().__eq__(other)
124             and self.path == other.path
125             and self.file_options.file_format == other.file_options.file_format
126             and self.file_options.s3_endpoint_override
127             == other.file_options.s3_endpoint_override
128         )
129 
130     @property
131     def path(self):
132         """
133         Returns the path of this file data source.
134         """
135         return self.file_options.uri
136 
137     @staticmethod
138     def from_proto(data_source: DataSourceProto):
139         return FileSource(
140             name=data_source.name,
141             field_mapping=dict(data_source.field_mapping),
142             file_format=FileFormat.from_proto(data_source.file_options.file_format),
143             path=data_source.file_options.uri,
144             timestamp_field=data_source.timestamp_field,
145             created_timestamp_column=data_source.created_timestamp_column,
146             s3_endpoint_override=data_source.file_options.s3_endpoint_override,
147             description=data_source.description,
148             tags=dict(data_source.tags),
149             owner=data_source.owner,
150         )
151 
152     def to_proto(self) -> DataSourceProto:
153         data_source_proto = DataSourceProto(
154             name=self.name,
155             type=DataSourceProto.BATCH_FILE,
156             field_mapping=self.field_mapping,
157             file_options=self.file_options.to_proto(),
158             description=self.description,
159             tags=self.tags,
160             owner=self.owner,
161         )
162 
163         data_source_proto.timestamp_field = self.timestamp_field
164         data_source_proto.created_timestamp_column = self.created_timestamp_column
165 
166         return data_source_proto
167 
168     def validate(self, config: RepoConfig):
169         # TODO: validate a FileSource
170         pass
171 
172     @staticmethod
173     def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:
174         return type_map.pa_to_feast_value_type
175 
176     def get_table_column_names_and_types(
177         self, config: RepoConfig
178     ) -> Iterable[Tuple[str, str]]:
179         filesystem, path = FileSource.create_filesystem_and_path(
180             self.path, self.file_options.s3_endpoint_override
181         )
182         schema = ParquetFile(
183             path if filesystem is None else filesystem.open_input_file(path)
184         ).schema_arrow
185         return zip(schema.names, map(str, schema.types))
186 
187     @staticmethod
188     def create_filesystem_and_path(
189         path: str, s3_endpoint_override: str
190     ) -> Tuple[Optional[FileSystem], str]:
191         if path.startswith("s3://"):
192             s3fs = S3FileSystem(
193                 endpoint_override=s3_endpoint_override if s3_endpoint_override else None
194             )
195             return s3fs, path.replace("s3://", "")
196         else:
197             return None, path
198 
199     def get_table_query_string(self) -> str:
200         pass
201 
202 
203 class FileOptions:
204     """
205     Configuration options for a file data source.
206     """
207 
208     def __init__(
209         self,
210         file_format: Optional[FileFormat],
211         s3_endpoint_override: Optional[str],
212         uri: Optional[str],
213     ):
214         """
215         Initializes a FileOptions object.
216 
217         Args:
218             file_format (optional): File source format, e.g. parquet.
219             s3_endpoint_override (optional): Custom s3 endpoint (used only with s3 uri).
220             uri (optional): File source url, e.g. s3:// or local file.
221         """
222         self.file_format = file_format
223         self.uri = uri or ""
224         self.s3_endpoint_override = s3_endpoint_override or ""
225 
226     @classmethod
227     def from_proto(cls, file_options_proto: DataSourceProto.FileOptions):
228         """
229         Creates a FileOptions from a protobuf representation of a file option
230 
231         Args:
232             file_options_proto: a protobuf representation of a datasource
233 
234         Returns:
235             Returns a FileOptions object based on the file_options protobuf
236         """
237         file_options = cls(
238             file_format=FileFormat.from_proto(file_options_proto.file_format),
239             uri=file_options_proto.uri,
240             s3_endpoint_override=file_options_proto.s3_endpoint_override,
241         )
242         return file_options
243 
244     def to_proto(self) -> DataSourceProto.FileOptions:
245         """
246         Converts an FileOptionsProto object to its protobuf representation.
247 
248         Returns:
249             FileOptionsProto protobuf
250         """
251         file_options_proto = DataSourceProto.FileOptions(
252             file_format=(
253                 None if self.file_format is None else self.file_format.to_proto()
254             ),
255             uri=self.uri,
256             s3_endpoint_override=self.s3_endpoint_override,
257         )
258 
259         return file_options_proto
260 
261 
262 class SavedDatasetFileStorage(SavedDatasetStorage):
263     _proto_attr_name = "file_storage"
264 
265     file_options: FileOptions
266 
267     def __init__(
268         self,
269         path: str,
270         file_format: FileFormat = ParquetFormat(),
271         s3_endpoint_override: Optional[str] = None,
272     ):
273         self.file_options = FileOptions(
274             file_format=file_format,
275             s3_endpoint_override=s3_endpoint_override,
276             uri=path,
277         )
278 
279     @staticmethod
280     def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:
281         file_options = FileOptions.from_proto(storage_proto.file_storage)
282         return SavedDatasetFileStorage(
283             path=file_options.uri,
284             file_format=file_options.file_format,
285             s3_endpoint_override=file_options.s3_endpoint_override,
286         )
287 
288     def to_proto(self) -> SavedDatasetStorageProto:
289         return SavedDatasetStorageProto(file_storage=self.file_options.to_proto())
290 
291     def to_data_source(self) -> DataSource:
292         return FileSource(
293             path=self.file_options.uri,
294             file_format=self.file_options.file_format,
295             s3_endpoint_override=self.file_options.s3_endpoint_override,
296         )
297 
298 
299 class FileLoggingDestination(LoggingDestination):
300     _proto_kind = "file_destination"
301 
302     path: str
303     s3_endpoint_override: str
304     partition_by: Optional[List[str]]
305 
306     def __init__(
307         self,
308         *,
309         path: str,
310         s3_endpoint_override="",
311         partition_by: Optional[List[str]] = None,
312     ):
313         self.path = path
314         self.s3_endpoint_override = s3_endpoint_override
315         self.partition_by = partition_by
316 
317     @classmethod
318     def from_proto(cls, config_proto: LoggingConfigProto) -> "LoggingDestination":
319         return FileLoggingDestination(
320             path=config_proto.file_destination.path,
321             s3_endpoint_override=config_proto.file_destination.s3_endpoint_override,
322             partition_by=list(config_proto.file_destination.partition_by)
323             if config_proto.file_destination.partition_by
324             else None,
325         )
326 
327     def to_proto(self) -> LoggingConfigProto:
328         return LoggingConfigProto(
329             file_destination=LoggingConfigProto.FileDestination(
330                 path=self.path,
331                 s3_endpoint_override=self.s3_endpoint_override,
332                 partition_by=self.partition_by,
333             )
334         )
335 
336     def to_data_source(self) -> DataSource:
337         return FileSource(
338             path=self.path,
339             file_format=ParquetFormat(),
340             s3_endpoint_override=self.s3_endpoint_override,
341         )
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/offline_stores/file_source.py b/sdk/python/feast/infra/offline_stores/file_source.py
--- a/sdk/python/feast/infra/offline_stores/file_source.py
+++ b/sdk/python/feast/infra/offline_stores/file_source.py
@@ -3,7 +3,7 @@
 
 from pyarrow._fs import FileSystem
 from pyarrow._s3fs import S3FileSystem
-from pyarrow.parquet import ParquetFile
+from pyarrow.parquet import ParquetDataset
 
 from feast import type_map
 from feast.data_format import FileFormat, ParquetFormat
@@ -179,9 +179,9 @@
         filesystem, path = FileSource.create_filesystem_and_path(
             self.path, self.file_options.s3_endpoint_override
         )
-        schema = ParquetFile(
+        schema = ParquetDataset(
             path if filesystem is None else filesystem.open_input_file(path)
-        ).schema_arrow
+        ).schema.to_arrow_schema()
         return zip(schema.names, map(str, schema.types))
 
     @staticmethod

{"golden_diff": "diff --git a/sdk/python/feast/infra/offline_stores/file_source.py b/sdk/python/feast/infra/offline_stores/file_source.py\n--- a/sdk/python/feast/infra/offline_stores/file_source.py\n+++ b/sdk/python/feast/infra/offline_stores/file_source.py\n@@ -3,7 +3,7 @@\n \n from pyarrow._fs import FileSystem\n from pyarrow._s3fs import S3FileSystem\n-from pyarrow.parquet import ParquetFile\n+from pyarrow.parquet import ParquetDataset\n \n from feast import type_map\n from feast.data_format import FileFormat, ParquetFormat\n@@ -179,9 +179,9 @@\n         filesystem, path = FileSource.create_filesystem_and_path(\n             self.path, self.file_options.s3_endpoint_override\n         )\n-        schema = ParquetFile(\n+        schema = ParquetDataset(\n             path if filesystem is None else filesystem.open_input_file(path)\n-        ).schema_arrow\n+        ).schema.to_arrow_schema()\n         return zip(schema.names, map(str, schema.types))\n \n     @staticmethod\n", "issue": "Parquet Schema Inference only supports File, not directory\nWhen using a FileSource that is in Parquet format, if the source happens to be a directory of partitioned Parquet files, the following lines throw an error:\r\n\r\nhttps://github.com/feast-dev/feast/blob/01d3568168bb9febb9fbda4988283b3886c32a31/sdk/python/feast/infra/offline_stores/file_source.py#L182-L184\r\n\r\n`OSError: Expected file path, but /home/ubuntu/project/data/driver_stats_partitioned is a directory`\r\n\r\nHow to replicate:\r\n\r\n1. Start with a demo feast project (`feast init`)\r\n2. Create a partitioned Parquet Dataset.  Use the following to create a dataset with only a single timestamp for inference\r\n```\r\nimport pyarrow.parquet as pq\r\ndf = pq.read_table(\"./data/driver_stats.parquet\")\r\ndf = df.drop([\"created\"])\r\npq.write_to_dataset(df, \"./data/driver_stats_partitioned\")\r\n```\r\n3. Update the file source in `example.py` to look like this:\r\n```\r\ndriver_hourly_stats = FileSource(\r\n    path=\"/home/ubuntu/cado-feast/feature_store/exciting_sunbeam/data/driver_stats_partitioned2\",\r\n)\r\n```\r\n\r\n4. Run `feast apply`\r\nFor now, I've been able to fix by updating the above lines to:\r\n```\r\nschema = ParquetDataset(\r\n    path if filesystem is None else filesystem.open_input_file(path)\r\n).schema.to_arrow_schema()\r\n```\n", "before_files": [{"content": "import warnings\nfrom typing import Callable, Dict, Iterable, List, Optional, Tuple\n\nfrom pyarrow._fs import FileSystem\nfrom pyarrow._s3fs import S3FileSystem\nfrom pyarrow.parquet import ParquetFile\n\nfrom feast import type_map\nfrom feast.data_format import FileFormat, ParquetFormat\nfrom feast.data_source import DataSource\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\nclass FileSource(DataSource):\n    def __init__(\n        self,\n        *args,\n        path: Optional[str] = None,\n        event_timestamp_column: Optional[str] = \"\",\n        file_format: Optional[FileFormat] = None,\n        created_timestamp_column: Optional[str] = \"\",\n        field_mapping: Optional[Dict[str, str]] = None,\n        date_partition_column: Optional[str] = \"\",\n        s3_endpoint_override: Optional[str] = None,\n        name: Optional[str] = \"\",\n        description: Optional[str] = \"\",\n        tags: Optional[Dict[str, str]] = None,\n        owner: Optional[str] = \"\",\n        timestamp_field: Optional[str] = \"\",\n    ):\n        \"\"\"Create a FileSource from a file containing feature data. Only Parquet format supported.\n\n        Args:\n\n            path: File path to file containing feature data. Must contain an event_timestamp column, entity columns and\n                feature columns.\n            event_timestamp_column(optional): (Deprecated) Event timestamp column used for point in time joins of feature values.\n            created_timestamp_column (optional): Timestamp column when row was created, used for deduplicating rows.\n            file_format (optional): Explicitly set the file format. Allows Feast to bypass inferring the file format.\n            field_mapping: A dictionary mapping of column names in this data source to feature names in a feature table\n                or view. Only used for feature columns, not entities or timestamp columns.\n            date_partition_column (optional): Timestamp column used for partitioning.\n            s3_endpoint_override (optional): Overrides AWS S3 enpoint with custom S3 storage\n            name (optional): Name for the file source. Defaults to the path.\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 file source, typically the email of the primary\n                maintainer.\n            timestamp_field (optional): Event timestamp foe;d used for point in time\n                joins of feature values.\n\n        Examples:\n            >>> from feast import FileSource\n            >>> file_source = FileSource(path=\"my_features.parquet\", timestamp_field=\"event_timestamp\")\n        \"\"\"\n        positional_attributes = [\"path\"]\n        _path = path\n        if args:\n            if args:\n                warnings.warn(\n                    (\n                        \"File Source parameters should be specified as a keyword argument instead of a positional arg.\"\n                        \"Feast 0.23+ will not support positional arguments to construct File sources\"\n                    ),\n                    DeprecationWarning,\n                )\n                if len(args) > len(positional_attributes):\n                    raise ValueError(\n                        f\"Only {', '.join(positional_attributes)} are allowed as positional args when defining \"\n                        f\"File sources, for backwards compatibility.\"\n                    )\n                if len(args) >= 1:\n                    _path = args[0]\n        if _path is None:\n            raise ValueError(\n                'No \"path\" argument provided. Please set \"path\" to the location of your file source.'\n            )\n        self.file_options = FileOptions(\n            file_format=file_format,\n            uri=_path,\n            s3_endpoint_override=s3_endpoint_override,\n        )\n\n        if date_partition_column:\n            warnings.warn(\n                (\n                    \"The argument 'date_partition_column' is not supported for File sources.\"\n                    \"It will be removed in Feast 0.23+\"\n                ),\n                DeprecationWarning,\n            )\n\n        super().__init__(\n            name=name if name else path,\n            event_timestamp_column=event_timestamp_column,\n            created_timestamp_column=created_timestamp_column,\n            field_mapping=field_mapping,\n            description=description,\n            tags=tags,\n            owner=owner,\n            timestamp_field=timestamp_field,\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, FileSource):\n            raise TypeError(\"Comparisons should only involve FileSource class objects.\")\n\n        return (\n            super().__eq__(other)\n            and self.path == other.path\n            and self.file_options.file_format == other.file_options.file_format\n            and self.file_options.s3_endpoint_override\n            == other.file_options.s3_endpoint_override\n        )\n\n    @property\n    def path(self):\n        \"\"\"\n        Returns the path of this file data source.\n        \"\"\"\n        return self.file_options.uri\n\n    @staticmethod\n    def from_proto(data_source: DataSourceProto):\n        return FileSource(\n            name=data_source.name,\n            field_mapping=dict(data_source.field_mapping),\n            file_format=FileFormat.from_proto(data_source.file_options.file_format),\n            path=data_source.file_options.uri,\n            timestamp_field=data_source.timestamp_field,\n            created_timestamp_column=data_source.created_timestamp_column,\n            s3_endpoint_override=data_source.file_options.s3_endpoint_override,\n            description=data_source.description,\n            tags=dict(data_source.tags),\n            owner=data_source.owner,\n        )\n\n    def to_proto(self) -> DataSourceProto:\n        data_source_proto = DataSourceProto(\n            name=self.name,\n            type=DataSourceProto.BATCH_FILE,\n            field_mapping=self.field_mapping,\n            file_options=self.file_options.to_proto(),\n            description=self.description,\n            tags=self.tags,\n            owner=self.owner,\n        )\n\n        data_source_proto.timestamp_field = self.timestamp_field\n        data_source_proto.created_timestamp_column = self.created_timestamp_column\n\n        return data_source_proto\n\n    def validate(self, config: RepoConfig):\n        # TODO: validate a FileSource\n        pass\n\n    @staticmethod\n    def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:\n        return type_map.pa_to_feast_value_type\n\n    def get_table_column_names_and_types(\n        self, config: RepoConfig\n    ) -> Iterable[Tuple[str, str]]:\n        filesystem, path = FileSource.create_filesystem_and_path(\n            self.path, self.file_options.s3_endpoint_override\n        )\n        schema = ParquetFile(\n            path if filesystem is None else filesystem.open_input_file(path)\n        ).schema_arrow\n        return zip(schema.names, map(str, schema.types))\n\n    @staticmethod\n    def create_filesystem_and_path(\n        path: str, s3_endpoint_override: str\n    ) -> Tuple[Optional[FileSystem], str]:\n        if path.startswith(\"s3://\"):\n            s3fs = S3FileSystem(\n                endpoint_override=s3_endpoint_override if s3_endpoint_override else None\n            )\n            return s3fs, path.replace(\"s3://\", \"\")\n        else:\n            return None, path\n\n    def get_table_query_string(self) -> str:\n        pass\n\n\nclass FileOptions:\n    \"\"\"\n    Configuration options for a file data source.\n    \"\"\"\n\n    def __init__(\n        self,\n        file_format: Optional[FileFormat],\n        s3_endpoint_override: Optional[str],\n        uri: Optional[str],\n    ):\n        \"\"\"\n        Initializes a FileOptions object.\n\n        Args:\n            file_format (optional): File source format, e.g. parquet.\n            s3_endpoint_override (optional): Custom s3 endpoint (used only with s3 uri).\n            uri (optional): File source url, e.g. s3:// or local file.\n        \"\"\"\n        self.file_format = file_format\n        self.uri = uri or \"\"\n        self.s3_endpoint_override = s3_endpoint_override or \"\"\n\n    @classmethod\n    def from_proto(cls, file_options_proto: DataSourceProto.FileOptions):\n        \"\"\"\n        Creates a FileOptions from a protobuf representation of a file option\n\n        Args:\n            file_options_proto: a protobuf representation of a datasource\n\n        Returns:\n            Returns a FileOptions object based on the file_options protobuf\n        \"\"\"\n        file_options = cls(\n            file_format=FileFormat.from_proto(file_options_proto.file_format),\n            uri=file_options_proto.uri,\n            s3_endpoint_override=file_options_proto.s3_endpoint_override,\n        )\n        return file_options\n\n    def to_proto(self) -> DataSourceProto.FileOptions:\n        \"\"\"\n        Converts an FileOptionsProto object to its protobuf representation.\n\n        Returns:\n            FileOptionsProto protobuf\n        \"\"\"\n        file_options_proto = DataSourceProto.FileOptions(\n            file_format=(\n                None if self.file_format is None else self.file_format.to_proto()\n            ),\n            uri=self.uri,\n            s3_endpoint_override=self.s3_endpoint_override,\n        )\n\n        return file_options_proto\n\n\nclass SavedDatasetFileStorage(SavedDatasetStorage):\n    _proto_attr_name = \"file_storage\"\n\n    file_options: FileOptions\n\n    def __init__(\n        self,\n        path: str,\n        file_format: FileFormat = ParquetFormat(),\n        s3_endpoint_override: Optional[str] = None,\n    ):\n        self.file_options = FileOptions(\n            file_format=file_format,\n            s3_endpoint_override=s3_endpoint_override,\n            uri=path,\n        )\n\n    @staticmethod\n    def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:\n        file_options = FileOptions.from_proto(storage_proto.file_storage)\n        return SavedDatasetFileStorage(\n            path=file_options.uri,\n            file_format=file_options.file_format,\n            s3_endpoint_override=file_options.s3_endpoint_override,\n        )\n\n    def to_proto(self) -> SavedDatasetStorageProto:\n        return SavedDatasetStorageProto(file_storage=self.file_options.to_proto())\n\n    def to_data_source(self) -> DataSource:\n        return FileSource(\n            path=self.file_options.uri,\n            file_format=self.file_options.file_format,\n            s3_endpoint_override=self.file_options.s3_endpoint_override,\n        )\n\n\nclass FileLoggingDestination(LoggingDestination):\n    _proto_kind = \"file_destination\"\n\n    path: str\n    s3_endpoint_override: str\n    partition_by: Optional[List[str]]\n\n    def __init__(\n        self,\n        *,\n        path: str,\n        s3_endpoint_override=\"\",\n        partition_by: Optional[List[str]] = None,\n    ):\n        self.path = path\n        self.s3_endpoint_override = s3_endpoint_override\n        self.partition_by = partition_by\n\n    @classmethod\n    def from_proto(cls, config_proto: LoggingConfigProto) -> \"LoggingDestination\":\n        return FileLoggingDestination(\n            path=config_proto.file_destination.path,\n            s3_endpoint_override=config_proto.file_destination.s3_endpoint_override,\n            partition_by=list(config_proto.file_destination.partition_by)\n            if config_proto.file_destination.partition_by\n            else None,\n        )\n\n    def to_proto(self) -> LoggingConfigProto:\n        return LoggingConfigProto(\n            file_destination=LoggingConfigProto.FileDestination(\n                path=self.path,\n                s3_endpoint_override=self.s3_endpoint_override,\n                partition_by=self.partition_by,\n            )\n        )\n\n    def to_data_source(self) -> DataSource:\n        return FileSource(\n            path=self.path,\n            file_format=ParquetFormat(),\n            s3_endpoint_override=self.s3_endpoint_override,\n        )\n", "path": "sdk/python/feast/infra/offline_stores/file_source.py"}], "after_files": [{"content": "import warnings\nfrom typing import Callable, Dict, Iterable, List, Optional, Tuple\n\nfrom pyarrow._fs import FileSystem\nfrom pyarrow._s3fs import S3FileSystem\nfrom pyarrow.parquet import ParquetDataset\n\nfrom feast import type_map\nfrom feast.data_format import FileFormat, ParquetFormat\nfrom feast.data_source import DataSource\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\nclass FileSource(DataSource):\n    def __init__(\n        self,\n        *args,\n        path: Optional[str] = None,\n        event_timestamp_column: Optional[str] = \"\",\n        file_format: Optional[FileFormat] = None,\n        created_timestamp_column: Optional[str] = \"\",\n        field_mapping: Optional[Dict[str, str]] = None,\n        date_partition_column: Optional[str] = \"\",\n        s3_endpoint_override: Optional[str] = None,\n        name: Optional[str] = \"\",\n        description: Optional[str] = \"\",\n        tags: Optional[Dict[str, str]] = None,\n        owner: Optional[str] = \"\",\n        timestamp_field: Optional[str] = \"\",\n    ):\n        \"\"\"Create a FileSource from a file containing feature data. Only Parquet format supported.\n\n        Args:\n\n            path: File path to file containing feature data. Must contain an event_timestamp column, entity columns and\n                feature columns.\n            event_timestamp_column(optional): (Deprecated) Event timestamp column used for point in time joins of feature values.\n            created_timestamp_column (optional): Timestamp column when row was created, used for deduplicating rows.\n            file_format (optional): Explicitly set the file format. Allows Feast to bypass inferring the file format.\n            field_mapping: A dictionary mapping of column names in this data source to feature names in a feature table\n                or view. Only used for feature columns, not entities or timestamp columns.\n            date_partition_column (optional): Timestamp column used for partitioning.\n            s3_endpoint_override (optional): Overrides AWS S3 enpoint with custom S3 storage\n            name (optional): Name for the file source. Defaults to the path.\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 file source, typically the email of the primary\n                maintainer.\n            timestamp_field (optional): Event timestamp foe;d used for point in time\n                joins of feature values.\n\n        Examples:\n            >>> from feast import FileSource\n            >>> file_source = FileSource(path=\"my_features.parquet\", timestamp_field=\"event_timestamp\")\n        \"\"\"\n        positional_attributes = [\"path\"]\n        _path = path\n        if args:\n            if args:\n                warnings.warn(\n                    (\n                        \"File Source parameters should be specified as a keyword argument instead of a positional arg.\"\n                        \"Feast 0.23+ will not support positional arguments to construct File sources\"\n                    ),\n                    DeprecationWarning,\n                )\n                if len(args) > len(positional_attributes):\n                    raise ValueError(\n                        f\"Only {', '.join(positional_attributes)} are allowed as positional args when defining \"\n                        f\"File sources, for backwards compatibility.\"\n                    )\n                if len(args) >= 1:\n                    _path = args[0]\n        if _path is None:\n            raise ValueError(\n                'No \"path\" argument provided. Please set \"path\" to the location of your file source.'\n            )\n        self.file_options = FileOptions(\n            file_format=file_format,\n            uri=_path,\n            s3_endpoint_override=s3_endpoint_override,\n        )\n\n        if date_partition_column:\n            warnings.warn(\n                (\n                    \"The argument 'date_partition_column' is not supported for File sources.\"\n                    \"It will be removed in Feast 0.23+\"\n                ),\n                DeprecationWarning,\n            )\n\n        super().__init__(\n            name=name if name else path,\n            event_timestamp_column=event_timestamp_column,\n            created_timestamp_column=created_timestamp_column,\n            field_mapping=field_mapping,\n            description=description,\n            tags=tags,\n            owner=owner,\n            timestamp_field=timestamp_field,\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, FileSource):\n            raise TypeError(\"Comparisons should only involve FileSource class objects.\")\n\n        return (\n            super().__eq__(other)\n            and self.path == other.path\n            and self.file_options.file_format == other.file_options.file_format\n            and self.file_options.s3_endpoint_override\n            == other.file_options.s3_endpoint_override\n        )\n\n    @property\n    def path(self):\n        \"\"\"\n        Returns the path of this file data source.\n        \"\"\"\n        return self.file_options.uri\n\n    @staticmethod\n    def from_proto(data_source: DataSourceProto):\n        return FileSource(\n            name=data_source.name,\n            field_mapping=dict(data_source.field_mapping),\n            file_format=FileFormat.from_proto(data_source.file_options.file_format),\n            path=data_source.file_options.uri,\n            timestamp_field=data_source.timestamp_field,\n            created_timestamp_column=data_source.created_timestamp_column,\n            s3_endpoint_override=data_source.file_options.s3_endpoint_override,\n            description=data_source.description,\n            tags=dict(data_source.tags),\n            owner=data_source.owner,\n        )\n\n    def to_proto(self) -> DataSourceProto:\n        data_source_proto = DataSourceProto(\n            name=self.name,\n            type=DataSourceProto.BATCH_FILE,\n            field_mapping=self.field_mapping,\n            file_options=self.file_options.to_proto(),\n            description=self.description,\n            tags=self.tags,\n            owner=self.owner,\n        )\n\n        data_source_proto.timestamp_field = self.timestamp_field\n        data_source_proto.created_timestamp_column = self.created_timestamp_column\n\n        return data_source_proto\n\n    def validate(self, config: RepoConfig):\n        # TODO: validate a FileSource\n        pass\n\n    @staticmethod\n    def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:\n        return type_map.pa_to_feast_value_type\n\n    def get_table_column_names_and_types(\n        self, config: RepoConfig\n    ) -> Iterable[Tuple[str, str]]:\n        filesystem, path = FileSource.create_filesystem_and_path(\n            self.path, self.file_options.s3_endpoint_override\n        )\n        schema = ParquetDataset(\n            path if filesystem is None else filesystem.open_input_file(path)\n        ).schema.to_arrow_schema()\n        return zip(schema.names, map(str, schema.types))\n\n    @staticmethod\n    def create_filesystem_and_path(\n        path: str, s3_endpoint_override: str\n    ) -> Tuple[Optional[FileSystem], str]:\n        if path.startswith(\"s3://\"):\n            s3fs = S3FileSystem(\n                endpoint_override=s3_endpoint_override if s3_endpoint_override else None\n            )\n            return s3fs, path.replace(\"s3://\", \"\")\n        else:\n            return None, path\n\n    def get_table_query_string(self) -> str:\n        pass\n\n\nclass FileOptions:\n    \"\"\"\n    Configuration options for a file data source.\n    \"\"\"\n\n    def __init__(\n        self,\n        file_format: Optional[FileFormat],\n        s3_endpoint_override: Optional[str],\n        uri: Optional[str],\n    ):\n        \"\"\"\n        Initializes a FileOptions object.\n\n        Args:\n            file_format (optional): File source format, e.g. parquet.\n            s3_endpoint_override (optional): Custom s3 endpoint (used only with s3 uri).\n            uri (optional): File source url, e.g. s3:// or local file.\n        \"\"\"\n        self.file_format = file_format\n        self.uri = uri or \"\"\n        self.s3_endpoint_override = s3_endpoint_override or \"\"\n\n    @classmethod\n    def from_proto(cls, file_options_proto: DataSourceProto.FileOptions):\n        \"\"\"\n        Creates a FileOptions from a protobuf representation of a file option\n\n        Args:\n            file_options_proto: a protobuf representation of a datasource\n\n        Returns:\n            Returns a FileOptions object based on the file_options protobuf\n        \"\"\"\n        file_options = cls(\n            file_format=FileFormat.from_proto(file_options_proto.file_format),\n            uri=file_options_proto.uri,\n            s3_endpoint_override=file_options_proto.s3_endpoint_override,\n        )\n        return file_options\n\n    def to_proto(self) -> DataSourceProto.FileOptions:\n        \"\"\"\n        Converts an FileOptionsProto object to its protobuf representation.\n\n        Returns:\n            FileOptionsProto protobuf\n        \"\"\"\n        file_options_proto = DataSourceProto.FileOptions(\n            file_format=(\n                None if self.file_format is None else self.file_format.to_proto()\n            ),\n            uri=self.uri,\n            s3_endpoint_override=self.s3_endpoint_override,\n        )\n\n        return file_options_proto\n\n\nclass SavedDatasetFileStorage(SavedDatasetStorage):\n    _proto_attr_name = \"file_storage\"\n\n    file_options: FileOptions\n\n    def __init__(\n        self,\n        path: str,\n        file_format: FileFormat = ParquetFormat(),\n        s3_endpoint_override: Optional[str] = None,\n    ):\n        self.file_options = FileOptions(\n            file_format=file_format,\n            s3_endpoint_override=s3_endpoint_override,\n            uri=path,\n        )\n\n    @staticmethod\n    def from_proto(storage_proto: SavedDatasetStorageProto) -> SavedDatasetStorage:\n        file_options = FileOptions.from_proto(storage_proto.file_storage)\n        return SavedDatasetFileStorage(\n            path=file_options.uri,\n            file_format=file_options.file_format,\n            s3_endpoint_override=file_options.s3_endpoint_override,\n        )\n\n    def to_proto(self) -> SavedDatasetStorageProto:\n        return SavedDatasetStorageProto(file_storage=self.file_options.to_proto())\n\n    def to_data_source(self) -> DataSource:\n        return FileSource(\n            path=self.file_options.uri,\n            file_format=self.file_options.file_format,\n            s3_endpoint_override=self.file_options.s3_endpoint_override,\n        )\n\n\nclass FileLoggingDestination(LoggingDestination):\n    _proto_kind = \"file_destination\"\n\n    path: str\n    s3_endpoint_override: str\n    partition_by: Optional[List[str]]\n\n    def __init__(\n        self,\n        *,\n        path: str,\n        s3_endpoint_override=\"\",\n        partition_by: Optional[List[str]] = None,\n    ):\n        self.path = path\n        self.s3_endpoint_override = s3_endpoint_override\n        self.partition_by = partition_by\n\n    @classmethod\n    def from_proto(cls, config_proto: LoggingConfigProto) -> \"LoggingDestination\":\n        return FileLoggingDestination(\n            path=config_proto.file_destination.path,\n            s3_endpoint_override=config_proto.file_destination.s3_endpoint_override,\n            partition_by=list(config_proto.file_destination.partition_by)\n            if config_proto.file_destination.partition_by\n            else None,\n        )\n\n    def to_proto(self) -> LoggingConfigProto:\n        return LoggingConfigProto(\n            file_destination=LoggingConfigProto.FileDestination(\n                path=self.path,\n                s3_endpoint_override=self.s3_endpoint_override,\n                partition_by=self.partition_by,\n            )\n        )\n\n    def to_data_source(self) -> DataSource:\n        return FileSource(\n            path=self.path,\n            file_format=ParquetFormat(),\n            s3_endpoint_override=self.s3_endpoint_override,\n        )\n", "path": "sdk/python/feast/infra/offline_stores/file_source.py"}]}