prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  PostgresToGoogleCloudStorageOperator - Custom schema mapping
  Version : 1.10.12
  
  I used PostgresToGoogleCloudStorageOperator to export the data and the schema file as well. But I saw a column on Postgres was `TIMESTAMP without time zone` but in BigQuery the auto-create table (via `GoogleCloudStorageToBigQueryOperator`) used the JSON schema file and created the table. When I checked the BQ table the data type was `TIMESTAMP`.
  
  For without timezone data, **`DATETIME`** would be the right choice. So can we manually MAP the data types during the schema file export? 
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `airflow/providers/google/cloud/transfers/postgres_to_gcs.py`
  Content:
  ```
  1 #
  2 # Licensed to the Apache Software Foundation (ASF) under one
  3 # or more contributor license agreements.  See the NOTICE file
  4 # distributed with this work for additional information
  5 # regarding copyright ownership.  The ASF licenses this file
  6 # to you under the Apache License, Version 2.0 (the
  7 # "License"); you may not use this file except in compliance
  8 # with the License.  You may obtain a copy of the License at
  9 #
  10 #   http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing,
  13 # software distributed under the License is distributed on an
  14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  15 # KIND, either express or implied.  See the License for the
  16 # specific language governing permissions and limitations
  17 # under the License.
  18 """PostgreSQL to GCS operator."""
  19 
  20 import datetime
  21 import json
  22 import time
  23 import uuid
  24 from decimal import Decimal
  25 from typing import Dict
  26 
  27 import pendulum
  28 
  29 from airflow.providers.google.cloud.transfers.sql_to_gcs import BaseSQLToGCSOperator
  30 from airflow.providers.postgres.hooks.postgres import PostgresHook
  31 
  32 
  33 class _PostgresServerSideCursorDecorator:
  34     """
  35     Inspired by `_PrestoToGCSPrestoCursorAdapter` to keep this consistent.
  36 
  37     Decorator for allowing description to be available for postgres cursor in case server side
  38     cursor is used. It doesn't provide other methods except those needed in BaseSQLToGCSOperator,
  39     which is more of a safety feature.
  40     """
  41 
  42     def __init__(self, cursor):
  43         self.cursor = cursor
  44         self.rows = []
  45         self.initialized = False
  46 
  47     def __iter__(self):
  48         return self
  49 
  50     def __next__(self):
  51         if self.rows:
  52             return self.rows.pop()
  53         else:
  54             self.initialized = True
  55             return next(self.cursor)
  56 
  57     @property
  58     def description(self):
  59         """Fetch first row to initialize cursor description when using server side cursor."""
  60         if not self.initialized:
  61             element = self.cursor.fetchone()
  62             if element is not None:
  63                 self.rows.append(element)
  64             self.initialized = True
  65         return self.cursor.description
  66 
  67 
  68 class PostgresToGCSOperator(BaseSQLToGCSOperator):
  69     """
  70     Copy data from Postgres to Google Cloud Storage in JSON or CSV format.
  71 
  72     :param postgres_conn_id: Reference to a specific Postgres hook.
  73     :param use_server_side_cursor: If server-side cursor should be used for querying postgres.
  74         For detailed info, check https://www.psycopg.org/docs/usage.html#server-side-cursors
  75     :param cursor_itersize: How many records are fetched at a time in case of server-side cursor.
  76     """
  77 
  78     ui_color = '#a0e08c'
  79 
  80     type_map = {
  81         1114: 'TIMESTAMP',
  82         1184: 'TIMESTAMP',
  83         1082: 'TIMESTAMP',
  84         1083: 'TIMESTAMP',
  85         1005: 'INTEGER',
  86         1007: 'INTEGER',
  87         1016: 'INTEGER',
  88         20: 'INTEGER',
  89         21: 'INTEGER',
  90         23: 'INTEGER',
  91         16: 'BOOLEAN',
  92         700: 'FLOAT',
  93         701: 'FLOAT',
  94         1700: 'FLOAT',
  95     }
  96 
  97     def __init__(
  98         self,
  99         *,
  100         postgres_conn_id='postgres_default',
  101         use_server_side_cursor=False,
  102         cursor_itersize=2000,
  103         **kwargs,
  104     ):
  105         super().__init__(**kwargs)
  106         self.postgres_conn_id = postgres_conn_id
  107         self.use_server_side_cursor = use_server_side_cursor
  108         self.cursor_itersize = cursor_itersize
  109 
  110     def _unique_name(self):
  111         return f"{self.dag_id}__{self.task_id}__{uuid.uuid4()}" if self.use_server_side_cursor else None
  112 
  113     def query(self):
  114         """Queries Postgres and returns a cursor to the results."""
  115         hook = PostgresHook(postgres_conn_id=self.postgres_conn_id)
  116         conn = hook.get_conn()
  117         cursor = conn.cursor(name=self._unique_name())
  118         cursor.execute(self.sql, self.parameters)
  119         if self.use_server_side_cursor:
  120             cursor.itersize = self.cursor_itersize
  121             return _PostgresServerSideCursorDecorator(cursor)
  122         return cursor
  123 
  124     def field_to_bigquery(self, field) -> Dict[str, str]:
  125         return {
  126             'name': field[0],
  127             'type': self.type_map.get(field[1], "STRING"),
  128             'mode': 'REPEATED' if field[1] in (1009, 1005, 1007, 1016) else 'NULLABLE',
  129         }
  130 
  131     def convert_type(self, value, schema_type):
  132         """
  133         Takes a value from Postgres, and converts it to a value that's safe for
  134         JSON/Google Cloud Storage/BigQuery. Dates are converted to UTC seconds.
  135         Decimals are converted to floats. Times are converted to seconds.
  136         """
  137         if isinstance(value, (datetime.datetime, datetime.date)):
  138             return pendulum.parse(value.isoformat()).float_timestamp
  139         if isinstance(value, datetime.time):
  140             formatted_time = time.strptime(str(value), "%H:%M:%S")
  141             return int(
  142                 datetime.timedelta(
  143                     hours=formatted_time.tm_hour, minutes=formatted_time.tm_min, seconds=formatted_time.tm_sec
  144                 ).total_seconds()
  145             )
  146         if isinstance(value, dict):
  147             return json.dumps(value)
  148         if isinstance(value, Decimal):
  149             return float(value)
  150         return value
  151
  ```
  --- 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/airflow/providers/google/cloud/transfers/postgres_to_gcs.py b/airflow/providers/google/cloud/transfers/postgres_to_gcs.py
  --- a/airflow/providers/google/cloud/transfers/postgres_to_gcs.py
  +++ b/airflow/providers/google/cloud/transfers/postgres_to_gcs.py
  @@ -78,10 +78,10 @@
       ui_color = '#a0e08c'
   
       type_map = {
  -        1114: 'TIMESTAMP',
  +        1114: 'DATETIME',
           1184: 'TIMESTAMP',
  -        1082: 'TIMESTAMP',
  -        1083: 'TIMESTAMP',
  +        1082: 'DATE',
  +        1083: 'TIME',
           1005: 'INTEGER',
           1007: 'INTEGER',
           1016: 'INTEGER',
  @@ -131,18 +131,24 @@
       def convert_type(self, value, schema_type):
           """
           Takes a value from Postgres, and converts it to a value that's safe for
  -        JSON/Google Cloud Storage/BigQuery. Dates are converted to UTC seconds.
  -        Decimals are converted to floats. Times are converted to seconds.
  +        JSON/Google Cloud Storage/BigQuery.
  +        Timezone aware Datetime are converted to UTC seconds.
  +        Unaware Datetime, Date and Time are converted to ISO formatted strings.
  +        Decimals are converted to floats.
           """
  -        if isinstance(value, (datetime.datetime, datetime.date)):
  -            return pendulum.parse(value.isoformat()).float_timestamp
  +        if isinstance(value, datetime.datetime):
  +            iso_format_value = value.isoformat()
  +            if value.tzinfo is None:
  +                return iso_format_value
  +            return pendulum.parse(iso_format_value).float_timestamp
  +        if isinstance(value, datetime.date):
  +            return value.isoformat()
           if isinstance(value, datetime.time):
               formatted_time = time.strptime(str(value), "%H:%M:%S")
  -            return int(
  -                datetime.timedelta(
  -                    hours=formatted_time.tm_hour, minutes=formatted_time.tm_min, seconds=formatted_time.tm_sec
  -                ).total_seconds()
  +            time_delta = datetime.timedelta(
  +                hours=formatted_time.tm_hour, minutes=formatted_time.tm_min, seconds=formatted_time.tm_sec
               )
  +            return str(time_delta)
           if isinstance(value, dict):
               return json.dumps(value)
           if isinstance(value, Decimal):
  
verification_info: {"golden_diff": "diff --git a/airflow/providers/google/cloud/transfers/postgres_to_gcs.py b/airflow/providers/google/cloud/transfers/postgres_to_gcs.py\n--- a/airflow/providers/google/cloud/transfers/postgres_to_gcs.py\n+++ b/airflow/providers/google/cloud/transfers/postgres_to_gcs.py\n@@ -78,10 +78,10 @@\n     ui_color = '#a0e08c'\n \n     type_map = {\n-        1114: 'TIMESTAMP',\n+        1114: 'DATETIME',\n         1184: 'TIMESTAMP',\n-        1082: 'TIMESTAMP',\n-        1083: 'TIMESTAMP',\n+        1082: 'DATE',\n+        1083: 'TIME',\n         1005: 'INTEGER',\n         1007: 'INTEGER',\n         1016: 'INTEGER',\n@@ -131,18 +131,24 @@\n     def convert_type(self, value, schema_type):\n         \"\"\"\n         Takes a value from Postgres, and converts it to a value that's safe for\n-        JSON/Google Cloud Storage/BigQuery. Dates are converted to UTC seconds.\n-        Decimals are converted to floats. Times are converted to seconds.\n+        JSON/Google Cloud Storage/BigQuery.\n+        Timezone aware Datetime are converted to UTC seconds.\n+        Unaware Datetime, Date and Time are converted to ISO formatted strings.\n+        Decimals are converted to floats.\n         \"\"\"\n-        if isinstance(value, (datetime.datetime, datetime.date)):\n-            return pendulum.parse(value.isoformat()).float_timestamp\n+        if isinstance(value, datetime.datetime):\n+            iso_format_value = value.isoformat()\n+            if value.tzinfo is None:\n+                return iso_format_value\n+            return pendulum.parse(iso_format_value).float_timestamp\n+        if isinstance(value, datetime.date):\n+            return value.isoformat()\n         if isinstance(value, datetime.time):\n             formatted_time = time.strptime(str(value), \"%H:%M:%S\")\n-            return int(\n-                datetime.timedelta(\n-                    hours=formatted_time.tm_hour, minutes=formatted_time.tm_min, seconds=formatted_time.tm_sec\n-                ).total_seconds()\n+            time_delta = datetime.timedelta(\n+                hours=formatted_time.tm_hour, minutes=formatted_time.tm_min, seconds=formatted_time.tm_sec\n             )\n+            return str(time_delta)\n         if isinstance(value, dict):\n             return json.dumps(value)\n         if isinstance(value, Decimal):\n", "issue": "PostgresToGoogleCloudStorageOperator - Custom schema mapping\nVersion : 1.10.12\r\n\r\nI used PostgresToGoogleCloudStorageOperator to export the data and the schema file as well. But I saw a column on Postgres was `TIMESTAMP without time zone` but in BigQuery the auto-create table (via `GoogleCloudStorageToBigQueryOperator`) used the JSON schema file and created the table. When I checked the BQ table the data type was `TIMESTAMP`.\r\n\r\nFor without timezone data, **`DATETIME`** would be the right choice. So can we manually MAP the data types during the schema file export? \n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"PostgreSQL to GCS operator.\"\"\"\n\nimport datetime\nimport json\nimport time\nimport uuid\nfrom decimal import Decimal\nfrom typing import Dict\n\nimport pendulum\n\nfrom airflow.providers.google.cloud.transfers.sql_to_gcs import BaseSQLToGCSOperator\nfrom airflow.providers.postgres.hooks.postgres import PostgresHook\n\n\nclass _PostgresServerSideCursorDecorator:\n    \"\"\"\n    Inspired by `_PrestoToGCSPrestoCursorAdapter` to keep this consistent.\n\n    Decorator for allowing description to be available for postgres cursor in case server side\n    cursor is used. It doesn't provide other methods except those needed in BaseSQLToGCSOperator,\n    which is more of a safety feature.\n    \"\"\"\n\n    def __init__(self, cursor):\n        self.cursor = cursor\n        self.rows = []\n        self.initialized = False\n\n    def __iter__(self):\n        return self\n\n    def __next__(self):\n        if self.rows:\n            return self.rows.pop()\n        else:\n            self.initialized = True\n            return next(self.cursor)\n\n    @property\n    def description(self):\n        \"\"\"Fetch first row to initialize cursor description when using server side cursor.\"\"\"\n        if not self.initialized:\n            element = self.cursor.fetchone()\n            if element is not None:\n                self.rows.append(element)\n            self.initialized = True\n        return self.cursor.description\n\n\nclass PostgresToGCSOperator(BaseSQLToGCSOperator):\n    \"\"\"\n    Copy data from Postgres to Google Cloud Storage in JSON or CSV format.\n\n    :param postgres_conn_id: Reference to a specific Postgres hook.\n    :param use_server_side_cursor: If server-side cursor should be used for querying postgres.\n        For detailed info, check https://www.psycopg.org/docs/usage.html#server-side-cursors\n    :param cursor_itersize: How many records are fetched at a time in case of server-side cursor.\n    \"\"\"\n\n    ui_color = '#a0e08c'\n\n    type_map = {\n        1114: 'TIMESTAMP',\n        1184: 'TIMESTAMP',\n        1082: 'TIMESTAMP',\n        1083: 'TIMESTAMP',\n        1005: 'INTEGER',\n        1007: 'INTEGER',\n        1016: 'INTEGER',\n        20: 'INTEGER',\n        21: 'INTEGER',\n        23: 'INTEGER',\n        16: 'BOOLEAN',\n        700: 'FLOAT',\n        701: 'FLOAT',\n        1700: 'FLOAT',\n    }\n\n    def __init__(\n        self,\n        *,\n        postgres_conn_id='postgres_default',\n        use_server_side_cursor=False,\n        cursor_itersize=2000,\n        **kwargs,\n    ):\n        super().__init__(**kwargs)\n        self.postgres_conn_id = postgres_conn_id\n        self.use_server_side_cursor = use_server_side_cursor\n        self.cursor_itersize = cursor_itersize\n\n    def _unique_name(self):\n        return f\"{self.dag_id}__{self.task_id}__{uuid.uuid4()}\" if self.use_server_side_cursor else None\n\n    def query(self):\n        \"\"\"Queries Postgres and returns a cursor to the results.\"\"\"\n        hook = PostgresHook(postgres_conn_id=self.postgres_conn_id)\n        conn = hook.get_conn()\n        cursor = conn.cursor(name=self._unique_name())\n        cursor.execute(self.sql, self.parameters)\n        if self.use_server_side_cursor:\n            cursor.itersize = self.cursor_itersize\n            return _PostgresServerSideCursorDecorator(cursor)\n        return cursor\n\n    def field_to_bigquery(self, field) -> Dict[str, str]:\n        return {\n            'name': field[0],\n            'type': self.type_map.get(field[1], \"STRING\"),\n            'mode': 'REPEATED' if field[1] in (1009, 1005, 1007, 1016) else 'NULLABLE',\n        }\n\n    def convert_type(self, value, schema_type):\n        \"\"\"\n        Takes a value from Postgres, and converts it to a value that's safe for\n        JSON/Google Cloud Storage/BigQuery. Dates are converted to UTC seconds.\n        Decimals are converted to floats. Times are converted to seconds.\n        \"\"\"\n        if isinstance(value, (datetime.datetime, datetime.date)):\n            return pendulum.parse(value.isoformat()).float_timestamp\n        if isinstance(value, datetime.time):\n            formatted_time = time.strptime(str(value), \"%H:%M:%S\")\n            return int(\n                datetime.timedelta(\n                    hours=formatted_time.tm_hour, minutes=formatted_time.tm_min, seconds=formatted_time.tm_sec\n                ).total_seconds()\n            )\n        if isinstance(value, dict):\n            return json.dumps(value)\n        if isinstance(value, Decimal):\n            return float(value)\n        return value\n", "path": "airflow/providers/google/cloud/transfers/postgres_to_gcs.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"PostgreSQL to GCS operator.\"\"\"\n\nimport datetime\nimport json\nimport time\nimport uuid\nfrom decimal import Decimal\nfrom typing import Dict\n\nimport pendulum\n\nfrom airflow.providers.google.cloud.transfers.sql_to_gcs import BaseSQLToGCSOperator\nfrom airflow.providers.postgres.hooks.postgres import PostgresHook\n\n\nclass _PostgresServerSideCursorDecorator:\n    \"\"\"\n    Inspired by `_PrestoToGCSPrestoCursorAdapter` to keep this consistent.\n\n    Decorator for allowing description to be available for postgres cursor in case server side\n    cursor is used. It doesn't provide other methods except those needed in BaseSQLToGCSOperator,\n    which is more of a safety feature.\n    \"\"\"\n\n    def __init__(self, cursor):\n        self.cursor = cursor\n        self.rows = []\n        self.initialized = False\n\n    def __iter__(self):\n        return self\n\n    def __next__(self):\n        if self.rows:\n            return self.rows.pop()\n        else:\n            self.initialized = True\n            return next(self.cursor)\n\n    @property\n    def description(self):\n        \"\"\"Fetch first row to initialize cursor description when using server side cursor.\"\"\"\n        if not self.initialized:\n            element = self.cursor.fetchone()\n            if element is not None:\n                self.rows.append(element)\n            self.initialized = True\n        return self.cursor.description\n\n\nclass PostgresToGCSOperator(BaseSQLToGCSOperator):\n    \"\"\"\n    Copy data from Postgres to Google Cloud Storage in JSON or CSV format.\n\n    :param postgres_conn_id: Reference to a specific Postgres hook.\n    :param use_server_side_cursor: If server-side cursor should be used for querying postgres.\n        For detailed info, check https://www.psycopg.org/docs/usage.html#server-side-cursors\n    :param cursor_itersize: How many records are fetched at a time in case of server-side cursor.\n    \"\"\"\n\n    ui_color = '#a0e08c'\n\n    type_map = {\n        1114: 'DATETIME',\n        1184: 'TIMESTAMP',\n        1082: 'DATE',\n        1083: 'TIME',\n        1005: 'INTEGER',\n        1007: 'INTEGER',\n        1016: 'INTEGER',\n        20: 'INTEGER',\n        21: 'INTEGER',\n        23: 'INTEGER',\n        16: 'BOOLEAN',\n        700: 'FLOAT',\n        701: 'FLOAT',\n        1700: 'FLOAT',\n    }\n\n    def __init__(\n        self,\n        *,\n        postgres_conn_id='postgres_default',\n        use_server_side_cursor=False,\n        cursor_itersize=2000,\n        **kwargs,\n    ):\n        super().__init__(**kwargs)\n        self.postgres_conn_id = postgres_conn_id\n        self.use_server_side_cursor = use_server_side_cursor\n        self.cursor_itersize = cursor_itersize\n\n    def _unique_name(self):\n        return f\"{self.dag_id}__{self.task_id}__{uuid.uuid4()}\" if self.use_server_side_cursor else None\n\n    def query(self):\n        \"\"\"Queries Postgres and returns a cursor to the results.\"\"\"\n        hook = PostgresHook(postgres_conn_id=self.postgres_conn_id)\n        conn = hook.get_conn()\n        cursor = conn.cursor(name=self._unique_name())\n        cursor.execute(self.sql, self.parameters)\n        if self.use_server_side_cursor:\n            cursor.itersize = self.cursor_itersize\n            return _PostgresServerSideCursorDecorator(cursor)\n        return cursor\n\n    def field_to_bigquery(self, field) -> Dict[str, str]:\n        return {\n            'name': field[0],\n            'type': self.type_map.get(field[1], \"STRING\"),\n            'mode': 'REPEATED' if field[1] in (1009, 1005, 1007, 1016) else 'NULLABLE',\n        }\n\n    def convert_type(self, value, schema_type):\n        \"\"\"\n        Takes a value from Postgres, and converts it to a value that's safe for\n        JSON/Google Cloud Storage/BigQuery.\n        Timezone aware Datetime are converted to UTC seconds.\n        Unaware Datetime, Date and Time are converted to ISO formatted strings.\n        Decimals are converted to floats.\n        \"\"\"\n        if isinstance(value, datetime.datetime):\n            iso_format_value = value.isoformat()\n            if value.tzinfo is None:\n                return iso_format_value\n            return pendulum.parse(iso_format_value).float_timestamp\n        if isinstance(value, datetime.date):\n            return value.isoformat()\n        if isinstance(value, datetime.time):\n            formatted_time = time.strptime(str(value), \"%H:%M:%S\")\n            time_delta = datetime.timedelta(\n                hours=formatted_time.tm_hour, minutes=formatted_time.tm_min, seconds=formatted_time.tm_sec\n            )\n            return str(time_delta)\n        if isinstance(value, dict):\n            return json.dumps(value)\n        if isinstance(value, Decimal):\n            return float(value)\n        return value\n", "path": "airflow/providers/google/cloud/transfers/postgres_to_gcs.py"}]}