prompt: 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()
  
  ```
  
golden_diff: 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, {})
               )
  
verification_info: {"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"}]}