We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Unable to load_file using parquet
Version: Astro 0.2.0
Python: 3.8, 3.9

Astro is unable to run the task `load_file` with a parquet file.

It raises the following exception:
```
  Traceback (most recent call last):
    File "pyarrow/io.pxi", line 1511, in pyarrow.lib.get_native_file
    File "/home/tati/.virtualenvs/astro-py38/lib/python3.8/site-packages/pyarrow/util.py", line 99, in _stringify_path
      raise TypeError("not a path-like object")
  TypeError: not a path-like object
  
  During handling of the above exception, another exception occurred:
  
  Traceback (most recent call last):
    File "pyarrow/io.pxi", line 1517, in pyarrow.lib.get_native_file
    File "pyarrow/io.pxi", line 729, in pyarrow.lib.PythonFile.__cinit__
  TypeError: binary file expected, got text file
  
    warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/astro/sql/operators/agnostic_load_file.py`
Content:
```
1 """
2 Copyright Astronomer, Inc.
3 
4 Licensed under the Apache License, Version 2.0 (the "License");
5 you may not use this file except in compliance with the License.
6 You may obtain a copy of the License at
7 
8     http://www.apache.org/licenses/LICENSE-2.0
9 
10 Unless required by applicable law or agreed to in writing, software
11 distributed under the License is distributed on an "AS IS" BASIS,
12 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 See the License for the specific language governing permissions and
14 limitations under the License.
15 """
16 
17 import os
18 from typing import Union
19 from urllib.parse import urlparse
20 
21 import pandas as pd
22 from airflow.hooks.base import BaseHook
23 from airflow.models import BaseOperator
24 from smart_open import open
25 
26 from astro.sql.table import Table, TempTable, create_table_name
27 from astro.utils.cloud_storage_creds import gcs_client, s3fs_creds
28 from astro.utils.load_dataframe import move_dataframe_to_sql
29 from astro.utils.schema_util import get_schema
30 from astro.utils.task_id_helper import get_task_id
31 
32 
33 class AgnosticLoadFile(BaseOperator):
34     """Load S3/local table to postgres/snowflake database.
35 
36     :param path: File path.
37     :type path: str
38     :param output_table_name: Name of table to create.
39     :type output_table_name: str
40     :param file_conn_id: Airflow connection id of input file (optional)
41     :type file_conn_id: str
42     :param output_conn_id: Database connection id.
43     :type output_conn_id: str
44     """
45 
46     def __init__(
47         self,
48         path,
49         output_table: Union[TempTable, Table],
50         file_conn_id="",
51         chunksize=None,
52         **kwargs,
53     ) -> None:
54         super().__init__(**kwargs)
55         self.output_table: Union[TempTable, Table] = output_table
56         self.path = path
57         self.chunksize = chunksize
58         self.file_conn_id = file_conn_id
59         self.kwargs = kwargs
60         self.output_table = output_table
61 
62     def execute(self, context):
63         """Loads csv/parquet table from local/S3/GCS with Pandas.
64 
65         Infers SQL database type based on connection then loads table to db.
66         """
67 
68         # Read file with Pandas load method based on `file_type` (S3 or local).
69         df = self._load_dataframe(self.path)
70 
71         # Retrieve conn type
72         conn = BaseHook.get_connection(self.output_table.conn_id)
73         if type(self.output_table) == TempTable:
74             self.output_table = self.output_table.to_table(
75                 create_table_name(context=context), get_schema()
76             )
77         else:
78             self.output_table.schema = self.output_table.schema or get_schema()
79         move_dataframe_to_sql(
80             output_table_name=self.output_table.table_name,
81             conn_id=self.output_table.conn_id,
82             database=self.output_table.database,
83             warehouse=self.output_table.warehouse,
84             schema=self.output_table.schema,
85             df=df,
86             conn_type=conn.conn_type,
87             user=conn.login,
88         )
89         self.log.info(f"returning table {self.output_table}")
90         return self.output_table
91 
92     @staticmethod
93     def validate_path(path):
94         """Validate a URL or local file path"""
95         try:
96             result = urlparse(path)
97             return all([result.scheme, result.netloc]) or os.path.isfile(path)
98         except:
99             return False
100 
101     def _load_dataframe(self, path):
102         """Read file with Pandas.
103 
104         Select method based on `file_type` (S3 or local).
105         """
106 
107         if not AgnosticLoadFile.validate_path(path):
108             raise ValueError("Invalid path: {}".format(path))
109 
110         file_type = path.split(".")[-1]
111         transport_params = {
112             "s3": s3fs_creds,
113             "gs": gcs_client,
114             "": lambda: None,
115         }[urlparse(path).scheme]()
116         deserialiser = {
117             "parquet": pd.read_parquet,
118             "csv": pd.read_csv,
119             "json": pd.read_json,
120             "ndjson": pd.read_json,
121         }
122         deserialiser_params = {"ndjson": {"lines": True}}
123         with open(path, transport_params=transport_params) as stream:
124             return deserialiser[file_type](
125                 stream, **deserialiser_params.get(file_type, {})
126             )
127 
128 
129 def load_file(
130     path,
131     output_table=None,
132     file_conn_id=None,
133     task_id=None,
134     **kwargs,
135 ):
136     """Convert AgnosticLoadFile into a function.
137 
138     Returns an XComArg object.
139 
140     :param path: File path.
141     :type path: str
142     :param output_table: Table to create
143     :type output_table: Table
144     :param file_conn_id: Airflow connection id of input file (optional)
145     :type file_conn_id: str
146     :param task_id: task id, optional.
147     :type task_id: str
148     """
149 
150     task_id = task_id if task_id is not None else get_task_id("load_file", path)
151 
152     return AgnosticLoadFile(
153         task_id=task_id,
154         path=path,
155         output_table=output_table,
156         file_conn_id=file_conn_id,
157         **kwargs,
158     ).output
159
```
--- 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/src/astro/sql/operators/agnostic_load_file.py b/src/astro/sql/operators/agnostic_load_file.py
--- a/src/astro/sql/operators/agnostic_load_file.py
+++ b/src/astro/sql/operators/agnostic_load_file.py
@@ -119,8 +119,11 @@
             "json": pd.read_json,
             "ndjson": pd.read_json,
         }
+        mode = {"parquet": "rb"}
         deserialiser_params = {"ndjson": {"lines": True}}
-        with open(path, transport_params=transport_params) as stream:
+        with open(
+            path, mode=mode.get(file_type, "r"), transport_params=transport_params
+        ) as stream:
             return deserialiser[file_type](
                 stream, **deserialiser_params.get(file_type, {})
             )

{"golden_diff": "diff --git a/src/astro/sql/operators/agnostic_load_file.py b/src/astro/sql/operators/agnostic_load_file.py\n--- a/src/astro/sql/operators/agnostic_load_file.py\n+++ b/src/astro/sql/operators/agnostic_load_file.py\n@@ -119,8 +119,11 @@\n             \"json\": pd.read_json,\n             \"ndjson\": pd.read_json,\n         }\n+        mode = {\"parquet\": \"rb\"}\n         deserialiser_params = {\"ndjson\": {\"lines\": True}}\n-        with open(path, transport_params=transport_params) as stream:\n+        with open(\n+            path, mode=mode.get(file_type, \"r\"), transport_params=transport_params\n+        ) as stream:\n             return deserialiser[file_type](\n                 stream, **deserialiser_params.get(file_type, {})\n             )\n", "issue": "Unable to load_file using parquet\nVersion: Astro 0.2.0\r\nPython: 3.8, 3.9\r\n\r\nAstro is unable to run the task `load_file` with a parquet file.\r\n\r\nIt raises the following exception:\r\n```\r\n  Traceback (most recent call last):\r\n    File \"pyarrow/io.pxi\", line 1511, in pyarrow.lib.get_native_file\r\n    File \"/home/tati/.virtualenvs/astro-py38/lib/python3.8/site-packages/pyarrow/util.py\", line 99, in _stringify_path\r\n      raise TypeError(\"not a path-like object\")\r\n  TypeError: not a path-like object\r\n  \r\n  During handling of the above exception, another exception occurred:\r\n  \r\n  Traceback (most recent call last):\r\n    File \"pyarrow/io.pxi\", line 1517, in pyarrow.lib.get_native_file\r\n    File \"pyarrow/io.pxi\", line 729, in pyarrow.lib.PythonFile.__cinit__\r\n  TypeError: binary file expected, got text file\r\n  \r\n    warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))\r\n```\n", "before_files": [{"content": "\"\"\"\nCopyright Astronomer, Inc.\n\nLicensed under the Apache License, Version 2.0 (the \"License\");\nyou may not use this file except in compliance with the License.\nYou may obtain a copy of the License at\n\n    http://www.apache.org/licenses/LICENSE-2.0\n\nUnless required by applicable law or agreed to in writing, software\ndistributed under the License is distributed on an \"AS IS\" BASIS,\nWITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\nSee the License for the specific language governing permissions and\nlimitations under the License.\n\"\"\"\n\nimport os\nfrom typing import Union\nfrom urllib.parse import urlparse\n\nimport pandas as pd\nfrom airflow.hooks.base import BaseHook\nfrom airflow.models import BaseOperator\nfrom smart_open import open\n\nfrom astro.sql.table import Table, TempTable, create_table_name\nfrom astro.utils.cloud_storage_creds import gcs_client, s3fs_creds\nfrom astro.utils.load_dataframe import move_dataframe_to_sql\nfrom astro.utils.schema_util import get_schema\nfrom astro.utils.task_id_helper import get_task_id\n\n\nclass AgnosticLoadFile(BaseOperator):\n    \"\"\"Load S3/local table to postgres/snowflake database.\n\n    :param path: File path.\n    :type path: str\n    :param output_table_name: Name of table to create.\n    :type output_table_name: str\n    :param file_conn_id: Airflow connection id of input file (optional)\n    :type file_conn_id: str\n    :param output_conn_id: Database connection id.\n    :type output_conn_id: str\n    \"\"\"\n\n    def __init__(\n        self,\n        path,\n        output_table: Union[TempTable, Table],\n        file_conn_id=\"\",\n        chunksize=None,\n        **kwargs,\n    ) -> None:\n        super().__init__(**kwargs)\n        self.output_table: Union[TempTable, Table] = output_table\n        self.path = path\n        self.chunksize = chunksize\n        self.file_conn_id = file_conn_id\n        self.kwargs = kwargs\n        self.output_table = output_table\n\n    def execute(self, context):\n        \"\"\"Loads csv/parquet table from local/S3/GCS with Pandas.\n\n        Infers SQL database type based on connection then loads table to db.\n        \"\"\"\n\n        # Read file with Pandas load method based on `file_type` (S3 or local).\n        df = self._load_dataframe(self.path)\n\n        # Retrieve conn type\n        conn = BaseHook.get_connection(self.output_table.conn_id)\n        if type(self.output_table) == TempTable:\n            self.output_table = self.output_table.to_table(\n                create_table_name(context=context), get_schema()\n            )\n        else:\n            self.output_table.schema = self.output_table.schema or get_schema()\n        move_dataframe_to_sql(\n            output_table_name=self.output_table.table_name,\n            conn_id=self.output_table.conn_id,\n            database=self.output_table.database,\n            warehouse=self.output_table.warehouse,\n            schema=self.output_table.schema,\n            df=df,\n            conn_type=conn.conn_type,\n            user=conn.login,\n        )\n        self.log.info(f\"returning table {self.output_table}\")\n        return self.output_table\n\n    @staticmethod\n    def validate_path(path):\n        \"\"\"Validate a URL or local file path\"\"\"\n        try:\n            result = urlparse(path)\n            return all([result.scheme, result.netloc]) or os.path.isfile(path)\n        except:\n            return False\n\n    def _load_dataframe(self, path):\n        \"\"\"Read file with Pandas.\n\n        Select method based on `file_type` (S3 or local).\n        \"\"\"\n\n        if not AgnosticLoadFile.validate_path(path):\n            raise ValueError(\"Invalid path: {}\".format(path))\n\n        file_type = path.split(\".\")[-1]\n        transport_params = {\n            \"s3\": s3fs_creds,\n            \"gs\": gcs_client,\n            \"\": lambda: None,\n        }[urlparse(path).scheme]()\n        deserialiser = {\n            \"parquet\": pd.read_parquet,\n            \"csv\": pd.read_csv,\n            \"json\": pd.read_json,\n            \"ndjson\": pd.read_json,\n        }\n        deserialiser_params = {\"ndjson\": {\"lines\": True}}\n        with open(path, transport_params=transport_params) as stream:\n            return deserialiser[file_type](\n                stream, **deserialiser_params.get(file_type, {})\n            )\n\n\ndef load_file(\n    path,\n    output_table=None,\n    file_conn_id=None,\n    task_id=None,\n    **kwargs,\n):\n    \"\"\"Convert AgnosticLoadFile into a function.\n\n    Returns an XComArg object.\n\n    :param path: File path.\n    :type path: str\n    :param output_table: Table to create\n    :type output_table: Table\n    :param file_conn_id: Airflow connection id of input file (optional)\n    :type file_conn_id: str\n    :param task_id: task id, optional.\n    :type task_id: str\n    \"\"\"\n\n    task_id = task_id if task_id is not None else get_task_id(\"load_file\", path)\n\n    return AgnosticLoadFile(\n        task_id=task_id,\n        path=path,\n        output_table=output_table,\n        file_conn_id=file_conn_id,\n        **kwargs,\n    ).output\n", "path": "src/astro/sql/operators/agnostic_load_file.py"}], "after_files": [{"content": "\"\"\"\nCopyright Astronomer, Inc.\n\nLicensed under the Apache License, Version 2.0 (the \"License\");\nyou may not use this file except in compliance with the License.\nYou may obtain a copy of the License at\n\n    http://www.apache.org/licenses/LICENSE-2.0\n\nUnless required by applicable law or agreed to in writing, software\ndistributed under the License is distributed on an \"AS IS\" BASIS,\nWITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\nSee the License for the specific language governing permissions and\nlimitations under the License.\n\"\"\"\n\nimport os\nfrom typing import Union\nfrom urllib.parse import urlparse\n\nimport pandas as pd\nfrom airflow.hooks.base import BaseHook\nfrom airflow.models import BaseOperator\nfrom smart_open import open\n\nfrom astro.sql.table import Table, TempTable, create_table_name\nfrom astro.utils.cloud_storage_creds import gcs_client, s3fs_creds\nfrom astro.utils.load_dataframe import move_dataframe_to_sql\nfrom astro.utils.schema_util import get_schema\nfrom astro.utils.task_id_helper import get_task_id\n\n\nclass AgnosticLoadFile(BaseOperator):\n    \"\"\"Load S3/local table to postgres/snowflake database.\n\n    :param path: File path.\n    :type path: str\n    :param output_table_name: Name of table to create.\n    :type output_table_name: str\n    :param file_conn_id: Airflow connection id of input file (optional)\n    :type file_conn_id: str\n    :param output_conn_id: Database connection id.\n    :type output_conn_id: str\n    \"\"\"\n\n    def __init__(\n        self,\n        path,\n        output_table: Union[TempTable, Table],\n        file_conn_id=\"\",\n        chunksize=None,\n        **kwargs,\n    ) -> None:\n        super().__init__(**kwargs)\n        self.output_table: Union[TempTable, Table] = output_table\n        self.path = path\n        self.chunksize = chunksize\n        self.file_conn_id = file_conn_id\n        self.kwargs = kwargs\n        self.output_table = output_table\n\n    def execute(self, context):\n        \"\"\"Loads csv/parquet table from local/S3/GCS with Pandas.\n\n        Infers SQL database type based on connection then loads table to db.\n        \"\"\"\n\n        # Read file with Pandas load method based on `file_type` (S3 or local).\n        df = self._load_dataframe(self.path)\n\n        # Retrieve conn type\n        conn = BaseHook.get_connection(self.output_table.conn_id)\n        if type(self.output_table) == TempTable:\n            self.output_table = self.output_table.to_table(\n                create_table_name(context=context), get_schema()\n            )\n        else:\n            self.output_table.schema = self.output_table.schema or get_schema()\n        move_dataframe_to_sql(\n            output_table_name=self.output_table.table_name,\n            conn_id=self.output_table.conn_id,\n            database=self.output_table.database,\n            warehouse=self.output_table.warehouse,\n            schema=self.output_table.schema,\n            df=df,\n            conn_type=conn.conn_type,\n            user=conn.login,\n        )\n        self.log.info(f\"returning table {self.output_table}\")\n        return self.output_table\n\n    @staticmethod\n    def validate_path(path):\n        \"\"\"Validate a URL or local file path\"\"\"\n        try:\n            result = urlparse(path)\n            return all([result.scheme, result.netloc]) or os.path.isfile(path)\n        except:\n            return False\n\n    def _load_dataframe(self, path):\n        \"\"\"Read file with Pandas.\n\n        Select method based on `file_type` (S3 or local).\n        \"\"\"\n\n        if not AgnosticLoadFile.validate_path(path):\n            raise ValueError(\"Invalid path: {}\".format(path))\n\n        file_type = path.split(\".\")[-1]\n        transport_params = {\n            \"s3\": s3fs_creds,\n            \"gs\": gcs_client,\n            \"\": lambda: None,\n        }[urlparse(path).scheme]()\n        deserialiser = {\n            \"parquet\": pd.read_parquet,\n            \"csv\": pd.read_csv,\n            \"json\": pd.read_json,\n            \"ndjson\": pd.read_json,\n        }\n        mode = {\"parquet\": \"rb\"}\n        deserialiser_params = {\"ndjson\": {\"lines\": True}}\n        with open(\n            path, mode=mode.get(file_type, \"r\"), transport_params=transport_params\n        ) as stream:\n            return deserialiser[file_type](\n                stream, **deserialiser_params.get(file_type, {})\n            )\n\n\ndef load_file(\n    path,\n    output_table=None,\n    file_conn_id=None,\n    task_id=None,\n    **kwargs,\n):\n    \"\"\"Convert AgnosticLoadFile into a function.\n\n    Returns an XComArg object.\n\n    :param path: File path.\n    :type path: str\n    :param output_table: Table to create\n    :type output_table: Table\n    :param file_conn_id: Airflow connection id of input file (optional)\n    :type file_conn_id: str\n    :param task_id: task id, optional.\n    :type task_id: str\n    \"\"\"\n\n    task_id = task_id if task_id is not None else get_task_id(\"load_file\", path)\n\n    return AgnosticLoadFile(\n        task_id=task_id,\n        path=path,\n        output_table=output_table,\n        file_conn_id=file_conn_id,\n        **kwargs,\n    ).output\n", "path": "src/astro/sql/operators/agnostic_load_file.py"}]}