Fix bigquery type error when export format is parquet (#16027)

Author: sunki-hong

airflow/providers/google/cloud/transfers/sql_to_gcs.py

@@ -20,7 +20,7 @@
 import json
 import warnings
 from tempfile import NamedTemporaryFile
-from typing import Optional, Sequence, Union
+from typing import Dict, Optional, Sequence, Union
 
 import pyarrow as pa
 import pyarrow.parquet as pq
@@ -278,7 +278,8 @@ def _convert_parquet_schema(self, cursor):
         }
 
         columns = [field[0] for field in cursor.description]
-        bq_types = [self.field_to_bigquery(field) for field in cursor.description]
+        bq_fields = [self.field_to_bigquery(field) for field in cursor.description]
+        bq_types = [bq_field.get('type') if bq_field is not None else None for bq_field in bq_fields]
         pq_types = [type_map.get(bq_type, pa.string()) for bq_type in bq_types]
         parquet_schema = pa.schema(zip(columns, pq_types))
         return parquet_schema
@@ -288,7 +289,7 @@ def query(self):
         """Execute DBAPI query."""
 
     @abc.abstractmethod
-    def field_to_bigquery(self, field):
+    def field_to_bigquery(self, field) -> Dict[str, str]:
         """Convert a DBAPI field to BigQuery schema format."""
 
     @abc.abstractmethod

tests/providers/google/cloud/transfers/test_sql_to_gcs.py

@@ -17,6 +17,7 @@
 
 import json
 import unittest
+from typing import Dict
 from unittest import mock
 from unittest.mock import MagicMock, Mock
 
@@ -62,8 +63,12 @@
 
 
 class DummySQLToGCSOperator(BaseSQLToGCSOperator):
-    def field_to_bigquery(self, field):
-        pass
+    def field_to_bigquery(self, field) -> Dict[str, str]:
+        return {
+            'name': field[0],
+            'type': 'STRING',
+            'mode': 'NULLABLE',
+        }
 
     def convert_type(self, value, schema_type):
         return 'convert_type_return_value'