Repository: googleapis/python-bigquery
Pull request #2126 (2025-02-04 13:19:42)

Feat: Adds foreign_type_info attribute to table class and adds unit tests.
This PR adds support for the `foreign_type_info` attribute on the Table class and the associated tests.



Patch:
diff --git a/google/cloud/bigquery/_helpers.py b/google/cloud/bigquery/_helpers.py
index ea47af28..d40217c4 100644
--- a/google/cloud/bigquery/_helpers.py
+++ b/google/cloud/bigquery/_helpers.py
@@ -978,11 +978,11 @@ def _build_resource_from_properties(obj, filter_fields):
     """
     partial = {}
     for filter_field in filter_fields:
-        api_field = obj._PROPERTY_TO_API_FIELD.get(filter_field)
+        api_field = _get_sub_prop(obj._PROPERTY_TO_API_FIELD, filter_field)
         if api_field is None and filter_field not in obj._properties:
             raise ValueError("No property %s" % filter_field)
         elif api_field is not None:
-            partial[api_field] = obj._properties.get(api_field)
+            _set_sub_prop(partial, api_field, _get_sub_prop(obj._properties, api_field))
         else:
             # allows properties that are not defined in the library
             # and properties that have the same name as API resource key
diff --git a/google/cloud/bigquery/_pandas_helpers.py b/google/cloud/bigquery/_pandas_helpers.py
index 4f70f6c2..0017d92c 100644
--- a/google/cloud/bigquery/_pandas_helpers.py
+++ b/google/cloud/bigquery/_pandas_helpers.py
@@ -12,7 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Shared helper functions for connecting BigQuery and pandas."""
+"""Shared helper functions for connecting BigQuery and pandas.
+
+NOTE: This module is DEPRECATED. Please make updates in the pandas-gbq package,
+instead. See: go/pandas-gbq-and-bigframes-redundancy and
+https://github.com/googleapis/python-bigquery-pandas/blob/main/pandas_gbq/schema/pandas_to_bigquery.py
+"""
 
 import concurrent.futures
 from datetime import datetime
@@ -40,6 +45,16 @@ except ImportError as exc:
 else:
     import numpy
 
+
+try:
+    import pandas_gbq.schema.pandas_to_bigquery  # type: ignore
+
+    pandas_gbq_import_exception = None
+except ImportError as exc:
+    pandas_gbq = None
+    pandas_gbq_import_exception = exc
+
+
 try:
     import db_dtypes  # type: ignore
 
@@ -445,6 +460,10 @@ def _first_array_valid(series):
 def dataframe_to_bq_schema(dataframe, bq_schema):
     """Convert a pandas DataFrame schema to a BigQuery schema.
 
+    DEPRECATED: Use
+    pandas_gbq.schema.pandas_to_bigquery.dataframe_to_bigquery_fields(),
+    instead. See: go/pandas-gbq-and-bigframes-redundancy.
+
     Args:
         dataframe (pandas.DataFrame):
             DataFrame for which the client determines the BigQuery schema.
@@ -460,6 +479,20 @@ def dataframe_to_bq_schema(dataframe, bq_schema):
             The automatically determined schema. Returns None if the type of
             any column cannot be determined.
     """
+    if pandas_gbq is None:
+        warnings.warn(
+            "Loading pandas DataFrame into BigQuery will require pandas-gbq "
+            "package version 0.26.1 or greater in the future. "
+            f"Tried to import pandas-gbq and got: {pandas_gbq_import_exception}",
+            category=FutureWarning,
+        )
+    else:
+        return pandas_gbq.schema.pandas_to_bigquery.dataframe_to_bigquery_fields(
+            dataframe,
+            override_bigquery_fields=bq_schema,
+            index=True,
+        )
+
     if bq_schema:
         bq_schema = schema._to_schema_fields(bq_schema)
         bq_schema_index = {field.name: field for field in bq_schema}
diff --git a/google/cloud/bigquery/_pyarrow_helpers.py b/google/cloud/bigquery/_pyarrow_helpers.py
index 3c745a61..1b42cd5c 100644
--- a/google/cloud/bigquery/_pyarrow_helpers.py
+++ b/google/cloud/bigquery/_pyarrow_helpers.py
@@ -12,7 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Shared helper functions for connecting BigQuery and pyarrow."""
+"""Shared helper functions for connecting BigQuery and pyarrow.
+
+NOTE: This module is DEPRECATED. Please make updates in the pandas-gbq package,
+instead. See: go/pandas-gbq-and-bigframes-redundancy and
+https://github.com/googleapis/python-bigquery-pandas/blob/main/pandas_gbq/schema/pyarrow_to_bigquery.py
+"""
 
 from typing import Any
 
diff --git a/google/cloud/bigquery/schema.py b/google/cloud/bigquery/schema.py
index 0f011a27..03cde830 100644
--- a/google/cloud/bigquery/schema.py
+++ b/google/cloud/bigquery/schema.py
@@ -15,10 +15,9 @@
 """Schemas for BigQuery tables / queries."""
 
 from __future__ import annotations
-import collections
 import enum
 import typing
-from typing import Any, cast, Dict, Iterable, Optional, Union
+from typing import Any, cast, Dict, Iterable, Optional, Union, Sequence
 
 from google.cloud.bigquery import _helpers
 from google.cloud.bigquery import standard_sql
@@ -489,6 +488,8 @@ def _parse_schema_resource(info):
         Optional[Sequence[google.cloud.bigquery.schema.SchemaField`]:
             A list of parsed fields, or ``None`` if no "fields" key found.
     """
+    if isinstance(info, list):
+        return [SchemaField.from_api_repr(f) for f in info]
     return [SchemaField.from_api_repr(f) for f in info.get("fields", ())]
 
 
@@ -501,40 +502,46 @@ def _build_schema_resource(fields):
     Returns:
         Sequence[Dict]: Mappings describing the schema of the supplied fields.
     """
-    return [field.to_api_repr() for field in fields]
+    if isinstance(fields, Sequence):
+        # Input is a Sequence (e.g. a list): Process and return a list of SchemaFields
+        return [field.to_api_repr() for field in fields]
+
+    else:
+        raise TypeError("Schema must be a Sequence (e.g. a list) or None.")
 
 
 def _to_schema_fields(schema):
-    """Coerce `schema` to a list of schema field instances.
+    """Coerces schema to a list of SchemaField instances while
+    preserving the original structure as much as possible.
 
     Args:
-        schema(Sequence[Union[ \
-            :class:`~google.cloud.bigquery.schema.SchemaField`, \
-            Mapping[str, Any] \
-        ]]):
-            Table schema to convert. If some items are passed as mappings,
-            their content must be compatible with
-            :meth:`~google.cloud.bigquery.schema.SchemaField.from_api_repr`.
+        schema (Sequence[Union[ \
+                   :class:`~google.cloud.bigquery.schema.SchemaField`, \
+                   Mapping[str, Any] \
+                       ]
+                   ]
+               )::
+            Table schema to convert. Can be a list of SchemaField
+            objects or mappings.
 
     Returns:
-        Sequence[:class:`~google.cloud.bigquery.schema.SchemaField`]
+        A list of SchemaField objects.
 
     Raises:
-        Exception: If ``schema`` is not a sequence, or if any item in the
-        sequence is not a :class:`~google.cloud.bigquery.schema.SchemaField`
-        instance or a compatible mapping representation of the field.
+        TypeError: If schema is not a Sequence.
     """
-    for field in schema:
-        if not isinstance(field, (SchemaField, collections.abc.Mapping)):
-            raise ValueError(
-                "Schema items must either be fields or compatible "
-                "mapping representations."
-            )
 
-    return [
-        field if isinstance(field, SchemaField) else SchemaField.from_api_repr(field)
-        for field in schema
-    ]
+    if isinstance(schema, Sequence):
+        # Input is a Sequence (e.g. a list): Process and return a list of SchemaFields
+        return [
+            field
+            if isinstance(field, SchemaField)
+            else SchemaField.from_api_repr(field)
+            for field in schema
+        ]
+
+    else:
+        raise TypeError("Schema must be a Sequence (e.g. a list) or None.")
 
 
 class PolicyTagList(object):
diff --git a/google/cloud/bigquery/table.py b/google/cloud/bigquery/table.py
index 934a28cf..c70a0ebe 100644
--- a/google/cloud/bigquery/table.py
+++ b/google/cloud/bigquery/table.py
@@ -21,7 +21,8 @@ import datetime
 import functools
 import operator
 import typing
-from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union, Sequence
+
 import warnings
 
 try:
@@ -66,6 +67,7 @@ from google.cloud.bigquery._tqdm_helpers import get_progress_bar
 from google.cloud.bigquery.encryption_configuration import EncryptionConfiguration
 from google.cloud.bigquery.enums import DefaultPandasDTypes
 from google.cloud.bigquery.external_config import ExternalConfig
+from google.cloud.bigquery import schema as _schema
 from google.cloud.bigquery.schema import _build_schema_resource
 from google.cloud.bigquery.schema import _parse_schema_resource
 from google.cloud.bigquery.schema import _to_schema_fields
@@ -398,7 +400,7 @@ class Table(_TableBase):
         "partitioning_type": "timePartitioning",
         "range_partitioning": "rangePartitioning",
         "time_partitioning": "timePartitioning",
-        "schema": "schema",
+        "schema": ["schema", "fields"],
         "snapshot_definition": "snapshotDefinition",
         "clone_definition": "cloneDefinition",
         "streaming_buffer": "streamingBuffer",
@@ -411,6 +413,7 @@ class Table(_TableBase):
         "max_staleness": "maxStaleness",
         "resource_tags": "resourceTags",
         "external_catalog_table_options": "externalCatalogTableOptions",
+        "foreign_type_info": ["schema", "foreignTypeInfo"],
     }
 
     def __init__(self, table_ref, schema=None) -> None:
@@ -451,8 +454,20 @@ class Table(_TableBase):
                 If ``schema`` is not a sequence, or if any item in the sequence
                 is not a :class:`~google.cloud.bigquery.schema.SchemaField`
                 instance or a compatible mapping representation of the field.
+
+        .. Note::
+            If you are referencing a schema for an external catalog table such
+            as a Hive table, it will also be necessary to populate the foreign_type_info
+            attribute. This is not necessary if defining the schema for a BigQuery table.
+
+            For details, see:
+            https://cloud.google.com/bigquery/docs/external-tables
+            https://cloud.google.com/bigquery/docs/datasets-intro#external_datasets
+
         """
-        prop = self._properties.get(self._PROPERTY_TO_API_FIELD["schema"])
+        prop = _helpers._get_sub_prop(
+            self._properties, self._PROPERTY_TO_API_FIELD["schema"]
+        )
         if not prop:
             return []
         else:
@@ -463,10 +478,21 @@ class Table(_TableBase):
         api_field = self._PROPERTY_TO_API_FIELD["schema"]
 
         if value is None:
-            self._properties[api_field] = None
-        else:
+            _helpers._set_sub_prop(
+                self._properties,
+                api_field,
+                None,
+            )
+        elif isinstance(value, Sequence):
             value = _to_schema_fields(value)
-            self._properties[api_field] = {"fields": _build_schema_resource(value)}
+            value = _build_schema_resource(value)
+            _helpers._set_sub_prop(
+                self._properties,
+                api_field,
+                value,
+            )
+        else:
+            raise TypeError("Schema must be a Sequence (e.g. a list) or None.")
 
     @property
     def labels(self):
@@ -1075,6 +1101,43 @@ class Table(_TableBase):
                 self._PROPERTY_TO_API_FIELD["external_catalog_table_options"]
             ] = value
 
+    @property
+    def foreign_type_info(self) -> Optional[_schema.ForeignTypeInfo]:
+        """Optional. Specifies metadata of the foreign data type definition in
+        field schema (TableFieldSchema.foreign_type_definition).
+
+        Returns:
+            Optional[schema.ForeignTypeInfo]:
+                Foreign type information, or :data:`None` if not set.
+
+        .. Note::
+            foreign_type_info is only required if you are referencing an
+            external catalog such as a Hive table.
+            For details, see:
+            https://cloud.google.com/bigquery/docs/external-tables
+            https://cloud.google.com/bigquery/docs/datasets-intro#external_datasets
+        """
+
+        prop = _helpers._get_sub_prop(
+            self._properties, self._PROPERTY_TO_API_FIELD["foreign_type_info"]
+        )
+        if prop is not None:
+            return _schema.ForeignTypeInfo.from_api_repr(prop)
+        return None
+
+    @foreign_type_info.setter
+    def foreign_type_info(self, value: Union[_schema.ForeignTypeInfo, dict, None]):
+        value = _helpers._isinstance_or_raise(
+            value,
+            (_schema.ForeignTypeInfo, dict),
+            none_allowed=True,
+        )
+        if isinstance(value, _schema.ForeignTypeInfo):
+            value = value.to_api_repr()
+        _helpers._set_sub_prop(
+            self._properties, self._PROPERTY_TO_API_FIELD["foreign_type_info"], value
+        )
+
     @classmethod
     def from_string(cls, full_table_id: str) -> "Table":
         """Construct a table from fully-qualified table ID.
diff --git a/noxfile.py b/noxfile.py
index e08956b1..87bd9a70 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -110,6 +110,14 @@ def default(session, install_extras=True):
     else:
         install_target = "."
     session.install("-e", install_target, "-c", constraints_path)
+
+    # Test with some broken "extras" in case the user didn't install the extra
+    # directly. For example, pandas-gbq is recommended for pandas features, but
+    # we want to test that we fallback to the previous behavior. For context,
+    # see internal document go/pandas-gbq-and-bigframes-redundancy.
+    if session.python == UNIT_TEST_PYTHON_VERSIONS[0]:
+        session.run("python", "-m", "pip", "uninstall", "pandas-gbq", "-y")
+
     session.run("python", "-m", "pip", "freeze")
 
     # Run py.test against the unit tests.
@@ -228,6 +236,13 @@ def system(session):
         extras = "[all]"
     session.install("-e", f".{extras}", "-c", constraints_path)
 
+    # Test with some broken "extras" in case the user didn't install the extra
+    # directly. For example, pandas-gbq is recommended for pandas features, but
+    # we want to test that we fallback to the previous behavior. For context,
+    # see internal document go/pandas-gbq-and-bigframes-redundancy.
+    if session.python == SYSTEM_TEST_PYTHON_VERSIONS[0]:
+        session.run("python", "-m", "pip", "uninstall", "pandas-gbq", "-y")
+
     # print versions of all dependencies
     session.run("python", "-m", "pip", "freeze")
 
diff --git a/pyproject.toml b/pyproject.toml
index ecf21d92..c4e5c2f0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -74,6 +74,9 @@ bqstorage = [
 ]
 pandas = [
   "pandas >= 1.1.0",
+  "pandas-gbq >= 0.26.1; python_version >= '3.8'",
+  "grpcio >= 1.47.0, < 2.0dev",
+  "grpcio >= 1.49.1, < 2.0dev; python_version >= '3.11'",
   "pyarrow >= 3.0.0",
   "db-dtypes >= 0.3.0, < 2.0.0dev",
   "importlib_metadata >= 1.0.0; python_version < '3.8'",


Test patch:
diff --git a/testing/constraints-3.8.txt b/testing/constraints-3.8.txt
index e5e73c5c..9883fb8c 100644
--- a/testing/constraints-3.8.txt
+++ b/testing/constraints-3.8.txt
@@ -1,2 +1,11 @@
 grpcio==1.47.0
 pandas==1.2.0
+
+# This constraints file is used to check that lower bounds
+# are correct in setup.py
+#
+# Pin the version to the lower bound.
+#
+# e.g., if setup.py has "foo >= 1.14.0, < 2.0.0dev",
+# Then this file should have foo==1.14.0
+pandas-gbq==0.26.1
diff --git a/tests/system/test_pandas.py b/tests/system/test_pandas.py
index 85c7b79e..a9e76d41 100644
--- a/tests/system/test_pandas.py
+++ b/tests/system/test_pandas.py
@@ -1259,7 +1259,7 @@ def test_upload_time_and_datetime_56(bigquery_client, dataset_id):
     df = pandas.DataFrame(
         dict(
             dt=[
-                datetime.datetime(2020, 1, 8, 8, 0, 0),
+                datetime.datetime(2020, 1, 8, 8, 0, 0, tzinfo=datetime.timezone.utc),
                 datetime.datetime(
                     2020,
                     1,
diff --git a/tests/unit/job/test_load.py b/tests/unit/job/test_load.py
index 0fb04469..10df46fb 100644
--- a/tests/unit/job/test_load.py
+++ b/tests/unit/job/test_load.py
@@ -272,7 +272,7 @@ class TestLoadJob(_Base):
 
         config = LoadJobConfig()
         full_name = SchemaField("full_name", "STRING", mode="REQUIRED")
-        with self.assertRaises(ValueError):
+        with self.assertRaises(TypeError):
             config.schema = [full_name, object()]
 
     def test_schema_setter(self):
diff --git a/tests/unit/test__pandas_helpers.py b/tests/unit/test__pandas_helpers.py
index edfaadf6..fdd232a5 100644
--- a/tests/unit/test__pandas_helpers.py
+++ b/tests/unit/test__pandas_helpers.py
@@ -35,6 +35,11 @@ try:
 except ImportError:
     pandas = None
 
+try:
+    import pandas_gbq.schema.pandas_to_bigquery
+except ImportError:
+    pandas_gbq = None
+
 try:
     import geopandas
 except ImportError:
@@ -1281,7 +1286,21 @@ def test_dataframe_to_parquet_compression_method(module_under_test):
 
 
 @pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
-def test_dataframe_to_bq_schema_w_named_index(module_under_test):
+@pytest.mark.skipif(pandas_gbq is None, reason="Requires `pandas-gbq`")
+def test_dataframe_to_bq_schema_returns_schema_with_pandas_gbq(
+    module_under_test, monkeypatch
+):
+    monkeypatch.setattr(module_under_test, "pandas_gbq", None)
+    dataframe = pandas.DataFrame({"field00": ["foo", "bar"]})
+    got = module_under_test.dataframe_to_bq_schema(dataframe, [])
+    # Don't assert beyond this, since pandas-gbq is now source of truth.
+    assert got is not None
+
+
+@pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
+def test_dataframe_to_bq_schema_w_named_index(module_under_test, monkeypatch):
+    monkeypatch.setattr(module_under_test, "pandas_gbq", None)
+
     df_data = collections.OrderedDict(
         [
             ("str_column", ["hello", "world"]),
@@ -1292,7 +1311,8 @@ def test_dataframe_to_bq_schema_w_named_index(module_under_test):
     index = pandas.Index(["a", "b"], name="str_index")
     dataframe = pandas.DataFrame(df_data, index=index)
 
-    returned_schema = module_under_test.dataframe_to_bq_schema(dataframe, [])
+    with pytest.warns(FutureWarning, match="pandas-gbq"):
+        returned_schema = module_under_test.dataframe_to_bq_schema(dataframe, [])
 
     expected_schema = (
         schema.SchemaField("str_index", "STRING", "NULLABLE"),
@@ -1304,7 +1324,9 @@ def test_dataframe_to_bq_schema_w_named_index(module_under_test):
 
 
 @pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
-def test_dataframe_to_bq_schema_w_multiindex(module_under_test):
+def test_dataframe_to_bq_schema_w_multiindex(module_under_test, monkeypatch):
+    monkeypatch.setattr(module_under_test, "pandas_gbq", None)
+
     df_data = collections.OrderedDict(
         [
             ("str_column", ["hello", "world"]),
@@ -1321,7 +1343,8 @@ def test_dataframe_to_bq_schema_w_multiindex(module_under_test):
     )
     dataframe = pandas.DataFrame(df_data, index=index)
 
-    returned_schema = module_under_test.dataframe_to_bq_schema(dataframe, [])
+    with pytest.warns(FutureWarning, match="pandas-gbq"):
+        returned_schema = module_under_test.dataframe_to_bq_schema(dataframe, [])
 
     expected_schema = (
         schema.SchemaField("str_index", "STRING", "NULLABLE"),
@@ -1335,7 +1358,9 @@ def test_dataframe_to_bq_schema_w_multiindex(module_under_test):
 
 
 @pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
-def test_dataframe_to_bq_schema_w_bq_schema(module_under_test):
+def test_dataframe_to_bq_schema_w_bq_schema(module_under_test, monkeypatch):
+    monkeypatch.setattr(module_under_test, "pandas_gbq", None)
+
     df_data = collections.OrderedDict(
         [
             ("str_column", ["hello", "world"]),
@@ -1350,7 +1375,10 @@ def test_dataframe_to_bq_schema_w_bq_schema(module_under_test):
         {"name": "bool_column", "type": "BOOL", "mode": "REQUIRED"},
     ]
 
-    returned_schema = module_under_test.dataframe_to_bq_schema(dataframe, dict_schema)
+    with pytest.warns(FutureWarning, match="pandas-gbq"):
+        returned_schema = module_under_test.dataframe_to_bq_schema(
+            dataframe, dict_schema
+        )
 
     expected_schema = (
         schema.SchemaField("str_column", "STRING", "NULLABLE"),
@@ -1361,7 +1389,11 @@ def test_dataframe_to_bq_schema_w_bq_schema(module_under_test):
 
 
 @pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
-def test_dataframe_to_bq_schema_fallback_needed_wo_pyarrow(module_under_test):
+def test_dataframe_to_bq_schema_fallback_needed_wo_pyarrow(
+    module_under_test, monkeypatch
+):
+    monkeypatch.setattr(module_under_test, "pandas_gbq", None)
+
     dataframe = pandas.DataFrame(
         data=[
             {"id": 10, "status": "FOO", "execution_date": datetime.date(2019, 5, 10)},
@@ -1389,7 +1421,11 @@ def test_dataframe_to_bq_schema_fallback_needed_wo_pyarrow(module_under_test):
 
 @pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
 @pytest.mark.skipif(isinstance(pyarrow, mock.Mock), reason="Requires `pyarrow`")
-def test_dataframe_to_bq_schema_fallback_needed_w_pyarrow(module_under_test):
+def test_dataframe_to_bq_schema_fallback_needed_w_pyarrow(
+    module_under_test, monkeypatch
+):
+    monkeypatch.setattr(module_under_test, "pandas_gbq", None)
+
     dataframe = pandas.DataFrame(
         data=[
             {"id": 10, "status": "FOO", "created_at": datetime.date(2019, 5, 10)},
@@ -1419,7 +1455,9 @@ def test_dataframe_to_bq_schema_fallback_needed_w_pyarrow(module_under_test):
 
 @pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
 @pytest.mark.skipif(isinstance(pyarrow, mock.Mock), reason="Requires `pyarrow`")
-def test_dataframe_to_bq_schema_pyarrow_fallback_fails(module_under_test):
+def test_dataframe_to_bq_schema_pyarrow_fallback_fails(module_under_test, monkeypatch):
+    monkeypatch.setattr(module_under_test, "pandas_gbq", None)
+
     dataframe = pandas.DataFrame(
         data=[
             {"struct_field": {"one": 2}, "status": "FOO"},
@@ -1443,9 +1481,11 @@ def test_dataframe_to_bq_schema_pyarrow_fallback_fails(module_under_test):
 
 
 @pytest.mark.skipif(geopandas is None, reason="Requires `geopandas`")
-def test_dataframe_to_bq_schema_geography(module_under_test):
+def test_dataframe_to_bq_schema_geography(module_under_test, monkeypatch):
     from shapely import wkt
 
+    monkeypatch.setattr(module_under_test, "pandas_gbq", None)
+
     df = geopandas.GeoDataFrame(
         pandas.DataFrame(
             dict(
@@ -1456,7 +1496,10 @@ def test_dataframe_to_bq_schema_geography(module_under_test):
         ),
         geometry="geo1",
     )
-    bq_schema = module_under_test.dataframe_to_bq_schema(df, [])
+
+    with pytest.warns(FutureWarning, match="pandas-gbq"):
+        bq_schema = module_under_test.dataframe_to_bq_schema(df, [])
+
     assert bq_schema == (
         schema.SchemaField("name", "STRING"),
         schema.SchemaField("geo1", "GEOGRAPHY"),
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index 462a70bb..6897c255 100644
--- a/tests/unit/test_client.py
+++ b/tests/unit/test_client.py
@@ -2051,7 +2051,7 @@ class TestClient(unittest.TestCase):
         ds.labels = LABELS
         ds.access_entries = [AccessEntry("OWNER", "userByEmail", "phred@example.com")]
         ds.resource_tags = RESOURCE_TAGS
-        fields = [
+        filter_fields = [
             "description",
             "friendly_name",
             "location",
@@ -2065,12 +2065,12 @@ class TestClient(unittest.TestCase):
         ) as final_attributes:
             ds2 = client.update_dataset(
                 ds,
-                fields=fields,
+                fields=filter_fields,
                 timeout=7.5,
             )
 
         final_attributes.assert_called_once_with(
-            {"path": "/%s" % PATH, "fields": fields}, client, None
+            {"path": "/%s" % PATH, "fields": filter_fields}, client, None
         )
 
         conn.api_request.assert_called_once_with(
@@ -2615,7 +2615,7 @@ class TestClient(unittest.TestCase):
         self.assertEqual(len(conn.api_request.call_args_list), 2)
         req = conn.api_request.call_args_list[1]
         self.assertEqual(req[1]["method"], "PATCH")
-        sent = {"schema": None}
+        sent = {"schema": {"fields": None}}
         self.assertEqual(req[1]["data"], sent)
         self.assertEqual(req[1]["path"], "/%s" % path)
         self.assertEqual(len(updated_table.schema), 0)
@@ -8391,8 +8391,12 @@ class TestClientUpload(object):
             autospec=True,
             side_effect=google.api_core.exceptions.NotFound("Table not found"),
         )
+        pandas_gbq_patch = mock.patch(
+            "google.cloud.bigquery._pandas_helpers.pandas_gbq",
+            new=None,
+        )
 
-        with load_patch as load_table_from_file, get_table_patch:
+        with load_patch as load_table_from_file, get_table_patch, pandas_gbq_patch:
             with warnings.catch_warnings(record=True) as warned:
                 client.load_table_from_dataframe(
                     dataframe, self.TABLE_REF, location=self.LOCATION
@@ -8448,7 +8452,6 @@ class TestClientUpload(object):
         load_patch = mock.patch(
             "google.cloud.bigquery.client.Client.load_table_from_file", autospec=True
         )
-
         get_table_patch = mock.patch(
             "google.cloud.bigquery.client.Client.get_table",
             autospec=True,
@@ -8460,6 +8463,7 @@ class TestClientUpload(object):
                 ]
             ),
         )
+
         with load_patch as load_table_from_file, get_table_patch:
             client.load_table_from_dataframe(
                 dataframe, self.TABLE_REF, location=self.LOCATION
@@ -8580,10 +8584,10 @@ class TestClientUpload(object):
 
         client = self._make_client()
         dataframe = pandas.DataFrame({"x": [1, 2, None, 4]}, dtype="Int64")
+
         load_patch = mock.patch(
             "google.cloud.bigquery.client.Client.load_table_from_file", autospec=True
         )
-
         get_table_patch = mock.patch(
             "google.cloud.bigquery.client.Client.get_table",
             autospec=True,
@@ -8612,8 +8616,11 @@ class TestClientUpload(object):
 
         sent_config = load_table_from_file.mock_calls[0][2]["job_config"]
         assert sent_config.source_format == job.SourceFormat.PARQUET
-        assert tuple(sent_config.schema) == (
-            SchemaField("x", "INT64", "NULLABLE", None),
+        assert (
+            # Accept either the GoogleSQL or legacy SQL type name from pandas-gbq.
+            tuple(sent_config.schema) == (SchemaField("x", "INT64", "NULLABLE", None),)
+            or tuple(sent_config.schema)
+            == (SchemaField("x", "INTEGER", "NULLABLE", None),)
         )
 
     def test_load_table_from_dataframe_struct_fields(self):
@@ -8759,7 +8766,7 @@ class TestClientUpload(object):
             data=records, columns=["float_column", "array_column"]
         )
 
-        expected_schema = [
+        expected_schema_googlesql = [
             SchemaField("float_column", "FLOAT"),
             SchemaField(
                 "array_column",
@@ -8767,6 +8774,14 @@ class TestClientUpload(object):
                 mode="REPEATED",
             ),
         ]
+        expected_schema_legacy_sql = [
+            SchemaField("float_column", "FLOAT"),
+            SchemaField(
+                "array_column",
+                "INTEGER",
+                mode="REPEATED",
+            ),
+        ]
 
         load_patch = mock.patch(
             "google.cloud.bigquery.client.Client.load_table_from_file", autospec=True
@@ -8802,7 +8817,10 @@ class TestClientUpload(object):
 
         sent_config = load_table_from_file.mock_calls[0][2]["job_config"]
         assert sent_config.source_format == job.SourceFormat.PARQUET
-        assert sent_config.schema == expected_schema
+        assert (
+            sent_config.schema == expected_schema_googlesql
+            or sent_config.schema == expected_schema_legacy_sql
+        )
 
     def test_load_table_from_dataframe_w_partial_schema(self):
         pandas = pytest.importorskip("pandas")
@@ -8922,7 +8940,6 @@ class TestClientUpload(object):
 
         load_table_from_file.assert_not_called()
         message = str(exc_context.value)
-        assert "bq_schema contains fields not present in dataframe" in message
         assert "unknown_col" in message
 
     def test_load_table_from_dataframe_w_schema_arrow_custom_compression(self):
diff --git a/tests/unit/test_schema.py b/tests/unit/test_schema.py
index 467f1e1d..3f2304a7 100644
--- a/tests/unit/test_schema.py
+++ b/tests/unit/test_schema.py
@@ -765,27 +765,62 @@ class Test_parse_schema_resource(unittest.TestCase, _SchemaBase):
         self._verifySchema(schema, RESOURCE)
 
 
-class Test_build_schema_resource(unittest.TestCase, _SchemaBase):
+class Test_build_schema_resource:
+    """Tests for the _build_schema_resource function."""
+
     def _call_fut(self, resource):
-        from google.cloud.bigquery.schema import _build_schema_resource
+        return schema._build_schema_resource(resource)
+
+    FULL_NAME = schema.SchemaField(
+        name="full_name", field_type="STRING", mode="REQUIRED"
+    )
+    AGE = schema.SchemaField(name="age", field_type="INTEGER", mode="REQUIRED")
+    LIST_RESOURCE = [
+        {"name": "full_name", "type": "STRING", "mode": "REQUIRED"},
+        {"name": "age", "type": "INTEGER", "mode": "REQUIRED"},
+    ]
+    FOREIGN_TYPE_INFO = schema.ForeignTypeInfo(type_system="TYPE_SYSTEM_UNSPECIFIED")
+    FOREIGN_TYPE_INFO_RESOURCE = {"typeSystem": "TYPE_SYSTEM_UNSPECIFIED"}
+
+    @pytest.mark.parametrize(
+        "schema,expected",
+        [
+            pytest.param([], [], id="empty list"),
+            pytest.param([FULL_NAME, AGE], LIST_RESOURCE, id="list"),
+        ],
+    )
+    def test_ctor_valid_input(self, schema, expected):
+        result = self._call_fut(schema)
+
+        assert result == expected
 
-        return _build_schema_resource(resource)
+    @pytest.mark.parametrize(
+        "schema,expected",
+        [
+            pytest.param(123, TypeError, id="invalid type"),
+        ],
+    )
+    def test_ctor_invalid_input(self, schema, expected):
+        with pytest.raises(TypeError) as e:
+            self._call_fut(schema)
+
+        # Looking for the first phrase from the string "Schema must be a ..."
+        assert "Schema must be a " in str(e.value)
 
     def test_defaults(self):
         from google.cloud.bigquery.schema import SchemaField
 
         full_name = SchemaField("full_name", "STRING", mode="REQUIRED")
         age = SchemaField("age", "INTEGER", mode="REQUIRED")
+        # test with simple list
         resource = self._call_fut([full_name, age])
-        self.assertEqual(len(resource), 2)
-        self.assertEqual(
-            resource[0],
-            {"name": "full_name", "type": "STRING", "mode": "REQUIRED"},
-        )
-        self.assertEqual(
-            resource[1],
-            {"name": "age", "type": "INTEGER", "mode": "REQUIRED"},
-        )
+        assert len(resource) == 2
+        assert resource[0] == {
+            "name": "full_name",
+            "type": "STRING",
+            "mode": "REQUIRED",
+        }
+        assert resource[1] == {"name": "age", "type": "INTEGER", "mode": "REQUIRED"}
 
     def test_w_description(self):
         from google.cloud.bigquery.schema import SchemaField
@@ -802,25 +837,20 @@ class Test_build_schema_resource(unittest.TestCase, _SchemaBase):
             description=None,
         )
         resource = self._call_fut([full_name, age])
-        self.assertEqual(len(resource), 2)
-        self.assertEqual(
-            resource[0],
-            {
-                "name": "full_name",
-                "type": "STRING",
-                "mode": "REQUIRED",
-                "description": DESCRIPTION,
-            },
-        )
-        self.assertEqual(
-            resource[1],
-            {
-                "name": "age",
-                "type": "INTEGER",
-                "mode": "REQUIRED",
-                "description": None,
-            },
-        )
+        assert len(resource) == 2
+        assert resource[0] == {
+            "name": "full_name",
+            "type": "STRING",
+            "mode": "REQUIRED",
+            "description": DESCRIPTION,
+        }
+
+        assert resource[1] == {
+            "name": "age",
+            "type": "INTEGER",
+            "mode": "REQUIRED",
+            "description": None,
+        }
 
     def test_w_subfields(self):
         from google.cloud.bigquery.schema import SchemaField
@@ -832,49 +862,72 @@ class Test_build_schema_resource(unittest.TestCase, _SchemaBase):
             "phone", "RECORD", mode="REPEATED", fields=[ph_type, ph_num]
         )
         resource = self._call_fut([full_name, phone])
-        self.assertEqual(len(resource), 2)
-        self.assertEqual(
-            resource[0],
-            {"name": "full_name", "type": "STRING", "mode": "REQUIRED"},
-        )
-        self.assertEqual(
-            resource[1],
-            {
-                "name": "phone",
-                "type": "RECORD",
-                "mode": "REPEATED",
-                "fields": [
-                    {"name": "type", "type": "STRING", "mode": "REQUIRED"},
-                    {"name": "number", "type": "STRING", "mode": "REQUIRED"},
-                ],
-            },
-        )
+        assert len(resource) == 2
+        assert resource[0] == {
+            "name": "full_name",
+            "type": "STRING",
+            "mode": "REQUIRED",
+        }
+        assert resource[1] == {
+            "name": "phone",
+            "type": "RECORD",
+            "mode": "REPEATED",
+            "fields": [
+                {"name": "type", "type": "STRING", "mode": "REQUIRED"},
+                {"name": "number", "type": "STRING", "mode": "REQUIRED"},
+            ],
+        }
+
 
+class Test_to_schema_fields:
+    """Tests for the _to_schema_fields function."""
 
-class Test_to_schema_fields(unittest.TestCase):
     @staticmethod
     def _call_fut(schema):
         from google.cloud.bigquery.schema import _to_schema_fields
 
         return _to_schema_fields(schema)
 
-    def test_invalid_type(self):
-        schema = [
-            ("full_name", "STRING", "REQUIRED"),
-            ("address", "STRING", "REQUIRED"),
-        ]
-        with self.assertRaises(ValueError):
-            self._call_fut(schema)
-
-    def test_schema_fields_sequence(self):
-        from google.cloud.bigquery.schema import SchemaField
+    FULL_NAME = schema.SchemaField(
+        name="full_name", field_type="STRING", mode="REQUIRED"
+    )
+    AGE = schema.SchemaField(name="age", field_type="INTEGER", mode="REQUIRED")
+    LIST_RESOURCE = [
+        {"name": "full_name", "type": "STRING", "mode": "REQUIRED"},
+        {"name": "age", "type": "INTEGER", "mode": "REQUIRED"},
+    ]
+    FOREIGN_TYPE_INFO = schema.ForeignTypeInfo(type_system="TYPE_SYSTEM_UNSPECIFIED")
+    FOREIGN_TYPE_INFO_RESOURCE = {"typeSystem": "TYPE_SYSTEM_UNSPECIFIED"}
 
-        schema = [
-            SchemaField("full_name", "STRING", mode="REQUIRED"),
-            SchemaField("age", "INT64", mode="NULLABLE"),
-        ]
+    @pytest.mark.parametrize(
+        "schema,expected",
+        [
+            pytest.param([], [], id="empty list"),
+            pytest.param((), [], id="empty tuple"),
+            pytest.param(LIST_RESOURCE, [FULL_NAME, AGE], id="list"),
+        ],
+    )
+    def test_ctor_valid_input(self, schema, expected):
         result = self._call_fut(schema)
-        self.assertEqual(result, schema)
+
+        assert result == expected
+
+    @pytest.mark.parametrize(
+        "schema,expected",
+        [
+            pytest.param(123, TypeError, id="invalid schema type"),
+            pytest.param([123, 123], TypeError, id="invalid SchemaField type"),
+            pytest.param({"fields": 123}, TypeError, id="invalid type, dict"),
+            pytest.param(
+                {"fields": 123, "foreignTypeInfo": 123},
+                TypeError,
+                id="invalid type, dict",
+            ),
+        ],
+    )
+    def test_ctor_invalid_input(self, schema, expected):
+        with pytest.raises(expected):
+            self._call_fut(schema)
 
     def test_unknown_properties(self):
         schema = [
@@ -933,7 +986,7 @@ class Test_to_schema_fields(unittest.TestCase):
         ]
 
         result = self._call_fut(schema)
-        self.assertEqual(result, expected_schema)
+        assert result == expected_schema
 
 
 class TestPolicyTags(unittest.TestCase):
diff --git a/tests/unit/test_table.py b/tests/unit/test_table.py
index 5154f01d..1a3d7ec0 100644
--- a/tests/unit/test_table.py
+++ b/tests/unit/test_table.py
@@ -31,6 +31,7 @@ from test_utils.imports import maybe_fail_import
 from google.cloud.bigquery import _versions_helpers
 from google.cloud.bigquery import exceptions
 from google.cloud.bigquery import external_config
+from google.cloud.bigquery import schema
 from google.cloud.bigquery.table import TableReference
 from google.cloud.bigquery.dataset import DatasetReference
 
@@ -699,7 +700,7 @@ class TestTable(unittest.TestCase, _SchemaBase):
         table_ref = dataset.table(self.TABLE_NAME)
         table = self._make_one(table_ref)
         full_name = SchemaField("full_name", "STRING", mode="REQUIRED")
-        with self.assertRaises(ValueError):
+        with self.assertRaises(TypeError):
             table.schema = [full_name, object()]
 
     def test_schema_setter_valid_fields(self):
@@ -1213,6 +1214,83 @@ class TestTable(unittest.TestCase, _SchemaBase):
         }
         self.assertEqual(resource, exp_resource)
 
+    def test_to_api_repr_w_schema_and_foreign_type_info(self):
+        """Tests to ensure that to_api_repr works correctly with
+        both schema and foreign_type_info fields
+        """
+
+        PROJECT = "test-project"
+        DATASET_ID = "test_dataset"
+        TABLE_ID = "coffee_table"
+        FOREIGNTYPEINFO = {
+            "typeSystem": "TYPE_SYSTEM_UNSPECIFIED",
+        }
+        SCHEMA = {
+            "fields": [
+                {"name": "full_name", "type": "STRING", "mode": "REQUIRED"},
+                {"name": "age", "type": "INTEGER", "mode": "REQUIRED"},
+            ],
+            "foreignTypeInfo": FOREIGNTYPEINFO,
+        }
+
+        API_REPR = {
+            "tableReference": {
+                "projectId": PROJECT,
+                "datasetId": DATASET_ID,
+                "tableId": TABLE_ID,
+            },
+            "schema": SCHEMA,
+        }
+
+        table = self._get_target_class().from_api_repr(API_REPR)
+        assert table._properties == table.to_api_repr()
+
+        # update schema (i.e. the fields), ensure foreign_type_info is unchanged
+        table.schema = []
+        expected = {
+            "fields": [],
+            "foreignTypeInfo": {"typeSystem": "TYPE_SYSTEM_UNSPECIFIED"},
+        }
+        assert table.to_api_repr()["schema"] == expected
+
+        # update foreign_type_info, ensure schema (i.e. the fields), is unchanged
+        table.foreign_type_info = {"typeSystem": "SCHEMA_SHOULD_NOT_CHANGE"}
+        expected = {
+            "fields": [],
+            "foreignTypeInfo": {"typeSystem": "SCHEMA_SHOULD_NOT_CHANGE"},
+        }
+        assert table.to_api_repr()["schema"] == expected
+
+    def test_from_api_repr_w_schema_and_foreign_type_info(self):
+        """Tests to ensure that to_api_repr works correctly with
+        both schema and foreign_type_info fields
+        """
+
+        PROJECT = "test-project"
+        DATASET_ID = "test_dataset"
+        TABLE_ID = "coffee_table"
+        FOREIGNTYPEINFO = {
+            "typeSystem": "TYPE_SYSTEM_UNSPECIFIED",
+        }
+        SCHEMA = {
+            "fields": [
+                {"name": "full_name", "type": "STRING", "mode": "REQUIRED"},
+                {"name": "age", "type": "INTEGER", "mode": "REQUIRED"},
+            ],
+            "foreignTypeInfo": FOREIGNTYPEINFO,
+        }
+        API_REPR = {
+            "tableReference": {
+                "projectId": PROJECT,
+                "datasetId": DATASET_ID,
+                "tableId": TABLE_ID,
+            },
+            "schema": SCHEMA,
+        }
+
+        table = self._get_target_class().from_api_repr(API_REPR)
+        assert table._properties == API_REPR
+
     def test__build_resource_w_custom_field(self):
         dataset = DatasetReference(self.PROJECT, self.DS_ID)
         table_ref = dataset.table(self.TABLE_NAME)
@@ -5993,6 +6071,99 @@ class TestExternalCatalogTableOptions:
         assert result == expected
 
 
+class TestForeignTypeInfo:
+    PROJECT = "test-project"
+    DATASET_ID = "test_dataset"
+    TABLE_ID = "coffee_table"
+    DATASET = DatasetReference(PROJECT, DATASET_ID)
+    TABLEREF = DATASET.table(TABLE_ID)
+    FOREIGNTYPEINFO = {
+        "typeSystem": "TYPE_SYSTEM_UNSPECIFIED",
+    }
+    API_REPR = {
+        "tableReference": {
+            "projectId": PROJECT,
+            "datasetId": DATASET_ID,
+            "tableId": TABLE_ID,
+        },
+        "schema": {
+            "fields": [
+                {"name": "full_name", "type": "STRING", "mode": "REQUIRED"},
+                {"name": "age", "type": "INTEGER", "mode": "REQUIRED"},
+            ],
+            "foreign_info_type": FOREIGNTYPEINFO,
+        },
+    }
+
+    from google.cloud.bigquery.schema import ForeignTypeInfo
+
+    @staticmethod
+    def _get_target_class(self):
+        from google.cloud.bigquery.table import Table
+
+        return Table
+
+    def _make_one(self, *args, **kw):
+        return self._get_target_class(self)(*args, **kw)
+
+    def test_foreign_type_info_default_initialization(self):
+        table = self._make_one(self.TABLEREF)
+        assert table.foreign_type_info is None
+
+    @pytest.mark.parametrize(
+        "foreign_type_info, expected",
+        [
+            (
+                {"typeSystem": "TYPE_SYSTEM_UNSPECIFIED"},
+                "TYPE_SYSTEM_UNSPECIFIED",
+            ),
+            (None, None),
+            (
+                ForeignTypeInfo(type_system="TYPE_SYSTEM_UNSPECIFIED"),
+                "TYPE_SYSTEM_UNSPECIFIED",
+            ),
+        ],
+    )
+    def test_foreign_type_info_valid_inputs(self, foreign_type_info, expected):
+        table = self._make_one(self.TABLEREF)
+
+        table.foreign_type_info = foreign_type_info
+
+        if foreign_type_info is None:
+            result = table.foreign_type_info
+        else:
+            result = table.foreign_type_info.type_system
+        assert result == expected
+
+    def test_foreign_type_info_invalid_inputs(self):
+        table = self._make_one(self.TABLEREF)
+
+        # invalid on the whole
+        with pytest.raises(TypeError, match="Pass .*"):
+            table.foreign_type_info = 123
+
+    def test_foreign_type_info_to_api_repr(self):
+        table = self._make_one(self.TABLEREF)
+
+        table.foreign_type_info = self.ForeignTypeInfo(
+            type_system="TYPE_SYSTEM_UNSPECIFIED",
+        )
+
+        result = table.to_api_repr()["schema"]["foreignTypeInfo"]
+        expected = self.FOREIGNTYPEINFO
+        assert result == expected
+
+    def test_foreign_type_info_from_api_repr(self):
+        table = self._make_one(self.TABLEREF)
+        table.foreign_type_info = self.FOREIGNTYPEINFO
+
+        fti = schema.ForeignTypeInfo.from_api_repr(self.FOREIGNTYPEINFO)
+
+        result = fti.to_api_repr()
+        expected = self.FOREIGNTYPEINFO
+        assert result == expected
+
+
 @pytest.mark.parametrize(
     "table_path",
     (