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()

```

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):

{"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"}]}