diff --git a/packages/google-cloud-bigquery/google/cloud/bigquery/_versions_helpers.py b/packages/google-cloud-bigquery/google/cloud/bigquery/_versions_helpers.py
index d856c19852e7..a9253448bc22 100644
--- a/packages/google-cloud-bigquery/google/cloud/bigquery/_versions_helpers.py
+++ b/packages/google-cloud-bigquery/google/cloud/bigquery/_versions_helpers.py
@@ -19,7 +19,6 @@
 
 from google.cloud.bigquery import exceptions
 
-
 _MIN_PYARROW_VERSION = packaging.version.Version("3.0.0")
 _MIN_BQ_STORAGE_VERSION = packaging.version.Version("2.0.0")
 _BQ_STORAGE_OPTIONAL_READ_SESSION_VERSION = packaging.version.Version("2.6.0")
@@ -247,3 +246,45 @@ def try_import(self, raise_if_error: bool = False) -> Any:
     and PYARROW_VERSIONS.try_import() is not None
     and PYARROW_VERSIONS.installed_version >= _MIN_PYARROW_VERSION_RANGE
 )
+
+
+class PandasGBQVersions:
+    """Version and delegation comparisons for pandas-gbq package."""
+
+    def __init__(self):
+        self._installed_version = None
+        self._delegation_api_version = None
+
+    @property
+    def installed_version(self) -> packaging.version.Version:
+        """Return the parsed version of pandas-gbq"""
+        if self._installed_version is not None:
+            return self._installed_version
+
+        try:
+            import pandas_gbq  # type: ignore
+
+            return packaging.version.parse(getattr(pandas_gbq, "__version__", "0.0.0"))
+        except Exception:
+            return packaging.version.parse("0.0.0")
+
+    @property
+    def delegation_api_version(self) -> int:
+        """Return the delegation API version of pandas-gbq if installed, otherwise 0."""
+        if self._delegation_api_version is not None:
+            return self._delegation_api_version
+
+        try:
+            import pandas_gbq  # type: ignore
+
+            return int(getattr(pandas_gbq, "_internal_delegation_api_version", 0))
+        except Exception:
+            return 0
+
+    @property
+    def is_delegation_supported(self) -> bool:
+        """True if the installed pandas-gbq version supports query delegation API (version >= 1)."""
+        return self.delegation_api_version >= 1
+
+
+PANDAS_GBQ_VERSIONS = PandasGBQVersions()
diff --git a/packages/google-cloud-bigquery/google/cloud/bigquery/dbapi/connection.py b/packages/google-cloud-bigquery/google/cloud/bigquery/dbapi/connection.py
index 53f32ce387ce..450e4188beef 100644
--- a/packages/google-cloud-bigquery/google/cloud/bigquery/dbapi/connection.py
+++ b/packages/google-cloud-bigquery/google/cloud/bigquery/dbapi/connection.py
@@ -78,10 +78,14 @@ def close(self):
         """
         self._closed = True
 
-        if self._owns_client:
+        for cursor_ in list(self._cursors_created):
+            if not cursor_._closed:
+                cursor_.close()
+
+        if self._owns_client and self._client is not None:
             self._client.close()
 
-        if self._owns_bqstorage_client:
+        if self._owns_bqstorage_client and self._bqstorage_client is not None:
             # There is no close() on the BQ Storage client itself.
             transport = self._bqstorage_client.transport
             transport.close()
@@ -95,10 +99,6 @@ def close(self):
             if channel is not None:
                 channel.close()
 
-        for cursor_ in self._cursors_created:
-            if not cursor_._closed:
-                cursor_.close()
-
     def commit(self):
         """No-op, but for consistency raise an error if connection is closed."""
 
diff --git a/packages/google-cloud-bigquery/google/cloud/bigquery/table.py b/packages/google-cloud-bigquery/google/cloud/bigquery/table.py
index 870cdcc5d2ab..9471a2aa6826 100644
--- a/packages/google-cloud-bigquery/google/cloud/bigquery/table.py
+++ b/packages/google-cloud-bigquery/google/cloud/bigquery/table.py
@@ -21,8 +21,7 @@
 import functools
 import operator
 import typing
-from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union, Sequence
-
+from typing import Any, Dict, Iterable, Iterator, List, Optional, Sequence, Tuple, Union
 import warnings
 
 try:
@@ -57,29 +56,34 @@
 
 import google.api_core.exceptions
 from google.api_core.page_iterator import HTTPIterator
-
 import google.cloud._helpers  # type: ignore
-from google.cloud.bigquery import _helpers
-from google.cloud.bigquery import _pandas_helpers
-from google.cloud.bigquery import _versions_helpers
+
+from google.cloud.bigquery import (
+    _helpers,
+    _pandas_helpers,
+    _string_references,
+    _versions_helpers,
+)
 from google.cloud.bigquery import exceptions as bq_exceptions
+from google.cloud.bigquery import external_config
+from google.cloud.bigquery import schema as _schema
 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
-from google.cloud.bigquery import external_config
-from google.cloud.bigquery import _string_references
+from google.cloud.bigquery.schema import (
+    _build_schema_resource,
+    _parse_schema_resource,
+    _to_schema_fields,
+)
 
 if typing.TYPE_CHECKING:  # pragma: NO COVER
     # Unconditionally import optional dependencies again to tell pytype that
     # they are not None, avoiding false "no attribute" errors.
+    import geopandas  # type: ignore
     import pandas
     import pyarrow
-    import geopandas  # type: ignore
+
     from google.cloud import bigquery_storage  # type: ignore
     from google.cloud.bigquery.dataset import DatasetReference
 
@@ -797,7 +801,7 @@ def time_partitioning(self, value):
             api_repr = value.to_api_repr()
         elif value is not None:
             raise ValueError(
-                "value must be google.cloud.bigquery.table.TimePartitioning " "or None"
+                "value must be google.cloud.bigquery.table.TimePartitioning or None"
             )
         self._properties[self._PROPERTY_TO_API_FIELD["time_partitioning"]] = api_repr
 
@@ -2801,6 +2805,24 @@ def to_dataframe(
             create_bqstorage_client = False
             bqstorage_client = None
 
+        if _versions_helpers.PANDAS_GBQ_VERSIONS.is_delegation_supported:
+            try:
+                import pandas_gbq  # type: ignore
+
+                pandas_gbq_version = getattr(pandas_gbq, "__version__", "0.0.0")
+            except ImportError:
+                pandas_gbq_version = "0.0.0"
+
+            client_info = getattr(
+                getattr(self.client, "_connection", None), "_client_info", None
+            )
+            if client_info:
+                ua = client_info.user_agent or ""
+                if "pandas-gbq" not in ua:
+                    client_info.user_agent = (
+                        f"{ua} pandas-gbq/{pandas_gbq_version}".strip()
+                    )
+
         record_batch = self.to_arrow(
             progress_bar_type=progress_bar_type,
             bqstorage_client=bqstorage_client,
@@ -2990,8 +3012,7 @@ def to_geodataframe(
         )
         if not geography_columns:
             raise TypeError(
-                "There must be at least one GEOGRAPHY column"
-                " to create a GeoDataFrame"
+                "There must be at least one GEOGRAPHY column to create a GeoDataFrame"
             )
 
         if geography_column:
diff --git a/packages/google-cloud-bigquery/tests/system/test_client.py b/packages/google-cloud-bigquery/tests/system/test_client.py
index 9ddec48428b1..0cc029c49530 100644
--- a/packages/google-cloud-bigquery/tests/system/test_client.py
+++ b/packages/google-cloud-bigquery/tests/system/test_client.py
@@ -204,12 +204,16 @@ def _still_in_use(bad_request):
             tag_key = key_values.pop()
 
             # Delete tag values first
-            [
-                tag_values_client.delete_tag_value(name=tag_value.name).result()
-                for tag_value in key_values
-            ]
+            for tag_value in key_values:
+                try:
+                    tag_values_client.delete_tag_value(name=tag_value.name).result()
+                except NotFound:
+                    pass
 
-            tag_keys_client.delete_tag_key(name=tag_key.name).result()
+            try:
+                tag_keys_client.delete_tag_key(name=tag_key.name).result()
+            except NotFound:
+                pass
 
     def test_get_service_account_email(self):
         client = Config.CLIENT
@@ -2184,6 +2188,7 @@ def test_dbapi_connection_does_not_leak_sockets(self):
         pytest.importorskip("google.cloud.bigquery_storage")
         current_process = psutil.Process()
         conn_start = current_process.net_connections()
+        conn_start_addrs = {c.laddr for c in conn_start if c.laddr}
         conn_count_start = len(conn_start)
 
         with helpers.patch_tracked_requests():
@@ -2202,24 +2207,29 @@ def test_dbapi_connection_does_not_leak_sockets(self):
             rows = cursor.fetchall()
             self.assertEqual(len(rows), 100000)
 
+            cursor.close()
             connection.close()
+
+        del connection, cursor, rows
         import gc
 
         gc.collect()
-        for _ in range(30):  # Wait up to 3 seconds
+        for _ in range(60):  # Wait up to 6 seconds for background socket cleanup
+            gc.collect()
             conn_end = current_process.net_connections()
             conn_count_end = len(conn_end)
-            if conn_count_end <= conn_count_start:
+            new_conns_remaining = [
+                c for c in conn_end if c.laddr and c.laddr not in conn_start_addrs
+            ]
+            if conn_count_end <= conn_count_start or len(new_conns_remaining) == 0:
                 break
             time.sleep(0.1)
 
         try:
-            self.assertLessEqual(conn_count_end, conn_count_start)
+            self.assertTrue(
+                conn_count_end <= conn_count_start or len(new_conns_remaining) == 0
+            )
         except AssertionError as e:
-            # Due to flakiness in this test (likely caused by OS cleanup delays or
-            # non-deterministic garbage collection of sockets), we want to capture
-            # the detailed state of connections in future failing runs to help
-            # decrease false positives and identify the root cause.
             conn_debug = [
                 f"Status: {c.status}, Laddr: {c.laddr}, Raddr: {c.raddr}"
                 for c in current_process.net_connections()
@@ -2231,6 +2241,7 @@ def test_dbapi_connection_does_not_leak_sockets(self):
                 f"--- Socket Leak Debug Info ---\n"
                 f"Start Count: {conn_count_start}\n"
                 f"End Count: {conn_count_end}\n"
+                f"New Sockets Remaining: {len(new_conns_remaining)}\n"
                 f"Current Connections:\n{debug_msg}"
             )
 
diff --git a/packages/google-cloud-bigquery/tests/unit/test__pandas_helpers.py b/packages/google-cloud-bigquery/tests/unit/test__pandas_helpers.py
index 34da6370e039..b22ad6850e44 100644
--- a/packages/google-cloud-bigquery/tests/unit/test__pandas_helpers.py
+++ b/packages/google-cloud-bigquery/tests/unit/test__pandas_helpers.py
@@ -18,6 +18,7 @@
 import decimal
 import functools
 import gc
+import importlib.metadata as metadata
 import operator
 import queue
 import time
@@ -25,8 +26,6 @@
 from unittest import mock
 import warnings
 
-import importlib.metadata as metadata
-
 try:
     import pandas
     import pandas.api.types
@@ -47,11 +46,12 @@
 import pytest
 
 from google import api_core
-
-from google.cloud.bigquery import exceptions
-from google.cloud.bigquery import _pyarrow_helpers
-from google.cloud.bigquery import _versions_helpers
-from google.cloud.bigquery import schema
+from google.cloud.bigquery import (
+    _pyarrow_helpers,
+    _versions_helpers,
+    exceptions,
+    schema,
+)
 from google.cloud.bigquery._pandas_helpers import determine_requested_streams
 
 pyarrow = _versions_helpers.PYARROW_VERSIONS.try_import()
@@ -1831,8 +1831,7 @@ def test__download_table_bqstorage(
     expected_call_count,
     expected_maxsize,
 ):
-    from google.cloud.bigquery import dataset
-    from google.cloud.bigquery import table
+    from google.cloud.bigquery import dataset, table
 
     queue_used = None  # A reference to the queue used by code under test.
 
@@ -1885,11 +1884,11 @@ def test__download_table_bqstorage_shuts_down_workers(
     the child threads are also stopped.
     """
     pytest.importorskip("google.cloud.bigquery_storage_v1")
-    from google.cloud.bigquery import dataset
-    from google.cloud.bigquery import table
     import google.cloud.bigquery_storage_v1.reader
     import google.cloud.bigquery_storage_v1.types
 
+    from google.cloud.bigquery import dataset, table
+
     monkeypatch.setattr(
         _versions_helpers.BQ_STORAGE_VERSIONS, "_installed_version", None
     )
@@ -2211,10 +2210,10 @@ def test_determine_requested_streams_invalid_max_stream_count():
     bigquery_storage is None, reason="Requires google-cloud-bigquery-storage"
 )
 def test__download_table_bqstorage_w_timeout_error(module_under_test):
-    from google.cloud.bigquery import dataset
-    from google.cloud.bigquery import table
     from unittest import mock
 
+    from google.cloud.bigquery import dataset, table
+
     mock_bqstorage_client = mock.create_autospec(
         bigquery_storage.BigQueryReadClient, instance=True
     )
@@ -2248,10 +2247,10 @@ def slow_download_stream(
     bigquery_storage is None, reason="Requires google-cloud-bigquery-storage"
 )
 def test__download_table_bqstorage_w_timeout_success(module_under_test):
-    from google.cloud.bigquery import dataset
-    from google.cloud.bigquery import table
     from unittest import mock
 
+    from google.cloud.bigquery import dataset, table
+
     mock_bqstorage_client = mock.create_autospec(
         bigquery_storage.BigQueryReadClient, instance=True
     )
@@ -2409,3 +2408,38 @@ def test_download_arrow_bqstorage_passes_timeout_to_create_read_session(
     assert retry_policy is not None
     # Check if deadline is set correctly in the retry policy
     assert retry_policy._deadline == timeout
+
+
+@pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
+def test_dataframe_to_bq_schema_w_unused_schema_field(module_under_test):
+    with mock.patch.object(module_under_test, "pandas_gbq", None):
+        with pytest.raises(
+            ValueError, match="bq_schema contains fields not present in dataframe"
+        ):
+            module_under_test.dataframe_to_bq_schema(
+                pandas.DataFrame(), (schema.SchemaField("not_in_df", "STRING"),)
+            )
+
+
+@pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
+@pytest.mark.skipif(isinstance(pyarrow, mock.Mock), reason="Requires `pyarrow`")
+def test_get_schema_by_pyarrow_bignumeric(module_under_test):
+    series = pandas.Series([decimal.Decimal("1.12345678901")])
+    result = module_under_test._get_schema_by_pyarrow("col", series)
+    assert result is not None
+    assert result.field_type == "BIGNUMERIC"
+
+
+@pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
+@pytest.mark.skipif(isinstance(pyarrow, mock.Mock), reason="Requires `pyarrow`")
+def test_get_types_mapper_range_timestamp_mismatch(module_under_test):
+    if not hasattr(pandas, "ArrowDtype"):
+        return
+    range_ts = pandas.ArrowDtype(
+        pyarrow.struct(
+            [("start", pyarrow.timestamp("us")), ("end", pyarrow.timestamp("us"))]
+        )
+    )
+    mapper = module_under_test.default_types_mapper(range_timestamp_dtype=range_ts)
+    unmatched_struct = pyarrow.struct([("other", pyarrow.int64())])
+    assert mapper(unmatched_struct) is None
diff --git a/packages/google-cloud-bigquery/tests/unit/test__pyarrow_helpers.py b/packages/google-cloud-bigquery/tests/unit/test__pyarrow_helpers.py
index c12a526de5d3..a4e9c0c9dc78 100644
--- a/packages/google-cloud-bigquery/tests/unit/test__pyarrow_helpers.py
+++ b/packages/google-cloud-bigquery/tests/unit/test__pyarrow_helpers.py
@@ -44,3 +44,16 @@ def test_bq_to_arrow_scalars(module_under_test):
 def test_arrow_scalar_ids_to_bq(module_under_test):
     assert module_under_test.arrow_scalar_ids_to_bq(pyarrow.bool_().id) == "BOOL"
     assert module_under_test.arrow_scalar_ids_to_bq("UNKNOWN_TYPE") is None
+
+
+def test_pyarrow_helpers_when_pyarrow_none(module_under_test):
+    import importlib
+    import sys
+    from unittest import mock
+
+    with mock.patch.dict(sys.modules, {"pyarrow": None}):
+        importlib.reload(module_under_test)
+        assert module_under_test.pyarrow is None
+        assert module_under_test.arrow_scalar_ids_to_bq(1) is None
+
+    importlib.reload(module_under_test)
diff --git a/packages/google-cloud-bigquery/tests/unit/test__versions_helpers.py b/packages/google-cloud-bigquery/tests/unit/test__versions_helpers.py
index 8379c87c18e0..06ce47104cb5 100644
--- a/packages/google-cloud-bigquery/tests/unit/test__versions_helpers.py
+++ b/packages/google-cloud-bigquery/tests/unit/test__versions_helpers.py
@@ -31,8 +31,7 @@
 except ImportError:
     pandas = None
 
-from google.cloud.bigquery import _versions_helpers
-from google.cloud.bigquery import exceptions
+from google.cloud.bigquery import _versions_helpers, exceptions
 
 
 @pytest.mark.skipif(pyarrow is None, reason="pyarrow is not installed")
@@ -59,14 +58,14 @@ def test_try_import_raises_error_w_legacy_pyarrow():
             versions.try_import(raise_if_error=True)
 
 
-@pytest.mark.skipif(
-    pyarrow is not None,
-    reason="pyarrow is installed, but this test needs it not to be",
-)
 def test_try_import_raises_error_w_no_pyarrow():
+    import sys
+
     versions = _versions_helpers.PyarrowVersions()
-    with pytest.raises(exceptions.LegacyPyarrowError):
-        versions.try_import(raise_if_error=True)
+    with mock.patch.dict(sys.modules, {"pyarrow": None}):
+        assert versions.try_import(raise_if_error=False) is None
+        with pytest.raises(exceptions.LegacyPyarrowError):
+            versions.try_import(raise_if_error=True)
 
 
 @pytest.mark.skipif(pyarrow is None, reason="pyarrow is not installed")
@@ -122,17 +121,29 @@ def test_returns_none_with_legacy_bqstorage():
         assert bq_storage is None
 
 
-@pytest.mark.skipif(
-    bigquery_storage is not None,
-    reason="Tests behavior when `google-cloud-bigquery-storage` isn't installed",
-)
 def test_returns_none_with_bqstorage_uninstalled():
-    try:
-        bqstorage_versions = _versions_helpers.BQStorageVersions()
-        bq_storage = bqstorage_versions.try_import()
-    except exceptions.LegacyBigQueryStorageError:  # pragma: NO COVER
-        raise ("NotFound error raised when raise_if_error == False.")
-    assert bq_storage is None
+    import sys
+
+    from google import cloud
+
+    versions = _versions_helpers.BQStorageVersions()
+    with mock.patch.dict(sys.modules, {"google.cloud.bigquery_storage": None}):
+        with mock.patch.dict(cloud.__dict__):
+            cloud.__dict__.pop("bigquery_storage", None)
+            assert versions.try_import() is None
+
+
+def test_raises_error_with_bqstorage_uninstalled():
+    import sys
+
+    from google import cloud
+
+    versions = _versions_helpers.BQStorageVersions()
+    with mock.patch.dict(sys.modules, {"google.cloud.bigquery_storage": None}):
+        with mock.patch.dict(cloud.__dict__):
+            cloud.__dict__.pop("bigquery_storage", None)
+            with pytest.raises(exceptions.BigQueryStorageNotFoundError):
+                versions.try_import(raise_if_error=True)
 
 
 @pytest.mark.skipif(
@@ -220,14 +231,14 @@ def test_try_import_raises_error_w_legacy_pandas():
             versions.try_import(raise_if_error=True)
 
 
-@pytest.mark.skipif(
-    pandas is not None,
-    reason="pandas is installed, but this test needs it not to be",
-)
 def test_try_import_raises_error_w_no_pandas():
+    import sys
+
     versions = _versions_helpers.PandasVersions()
-    with pytest.raises(exceptions.LegacyPandasError):
-        versions.try_import(raise_if_error=True)
+    with mock.patch.dict(sys.modules, {"pandas": None}):
+        assert versions.try_import(raise_if_error=False) is None
+        with pytest.raises(exceptions.LegacyPandasError):
+            versions.try_import(raise_if_error=True)
 
 
 @pytest.mark.skipif(pandas is None, reason="pandas is not installed")
@@ -246,3 +257,108 @@ def test_installed_pandas_version_returns_parsed_version():
     assert version.major == 1
     assert version.minor == 1
     assert version.micro == 0
+
+
+def test_installed_pandas_gbq_version_returns_cached():
+    versions = _versions_helpers.PandasGBQVersions()
+    versions._installed_version = object()
+    assert versions.installed_version is versions._installed_version
+
+
+def test_installed_pandas_gbq_version_returns_parsed_version():
+    import sys
+
+    mock_pandas_gbq = mock.Mock()
+    mock_pandas_gbq.__version__ = "1.2.3"
+    versions = _versions_helpers.PandasGBQVersions()
+    with mock.patch.dict(sys.modules, {"pandas_gbq": mock_pandas_gbq}):
+        version = versions.installed_version
+
+    assert version.major == 1
+    assert version.minor == 2
+    assert version.micro == 3
+
+
+def test_installed_pandas_gbq_version_falls_back_on_import_error():
+    import sys
+
+    versions = _versions_helpers.PandasGBQVersions()
+    with mock.patch.dict(sys.modules, {"pandas_gbq": None}):
+        version = versions.installed_version
+
+    assert version.major == 0
+    assert version.minor == 0
+    assert version.micro == 0
+
+
+def test_installed_pandas_gbq_version_falls_back_on_other_error():
+    import sys
+
+    # Simulate a corrupted package raising an error on import/property access
+    class CorruptPandasGBQ:
+        @property
+        def __version__(self):
+            raise TypeError("Corrupted package")
+
+    versions = _versions_helpers.PandasGBQVersions()
+    with mock.patch.dict(sys.modules, {"pandas_gbq": CorruptPandasGBQ()}):
+        version = versions.installed_version
+
+    assert version.major == 0
+    assert version.minor == 0
+    assert version.micro == 0
+
+
+def test_pandas_gbq_delegation_api_version_returns_cached():
+    versions = _versions_helpers.PandasGBQVersions()
+    versions._delegation_api_version = object()
+    assert versions.delegation_api_version is versions._delegation_api_version
+
+
+def test_pandas_gbq_delegation_api_version_returns_value():
+    import sys
+
+    mock_pandas_gbq = mock.Mock()
+    mock_pandas_gbq._internal_delegation_api_version = 42
+    versions = _versions_helpers.PandasGBQVersions()
+    with mock.patch.dict(sys.modules, {"pandas_gbq": mock_pandas_gbq}):
+        version = versions.delegation_api_version
+
+    assert version == 42
+
+
+def test_pandas_gbq_delegation_api_version_falls_back_on_import_error():
+    import sys
+
+    versions = _versions_helpers.PandasGBQVersions()
+    with mock.patch.dict(sys.modules, {"pandas_gbq": None}):
+        version = versions.delegation_api_version
+
+    assert version == 0
+
+
+def test_pandas_gbq_delegation_api_version_falls_back_on_other_error():
+    import sys
+
+    class CorruptPandasGBQ:
+        @property
+        def _internal_delegation_api_version(self):
+            raise TypeError("Corrupted package")
+
+    versions = _versions_helpers.PandasGBQVersions()
+    with mock.patch.dict(sys.modules, {"pandas_gbq": CorruptPandasGBQ()}):
+        version = versions.delegation_api_version
+
+    assert version == 0
+
+
+def test_pandas_gbq_is_delegation_supported_true():
+    versions = _versions_helpers.PandasGBQVersions()
+    versions._delegation_api_version = 1
+    assert versions.is_delegation_supported is True
+
+
+def test_pandas_gbq_is_delegation_supported_false():
+    versions = _versions_helpers.PandasGBQVersions()
+    versions._delegation_api_version = 0
+    assert versions.is_delegation_supported is False
diff --git a/packages/google-cloud-bigquery/tests/unit/test_magics.py b/packages/google-cloud-bigquery/tests/unit/test_magics.py
index 03a3a2dbbdba..e2eed9b35793 100644
--- a/packages/google-cloud-bigquery/tests/unit/test_magics.py
+++ b/packages/google-cloud-bigquery/tests/unit/test_magics.py
@@ -37,13 +37,13 @@
 
 bigquery_storage = pytest.importorskip("google.cloud.bigquery_storage")
 IPython = pytest.importorskip("IPython")
-interactiveshell = pytest.importorskip("IPython.terminal.interactiveshell")
+interactiveshell = pytest.importorskip("IPython.core.interactiveshell")
 tools = pytest.importorskip("IPython.testing.tools")
 io = pytest.importorskip("IPython.utils.io")
 pandas = pytest.importorskip("pandas")
 
 
-@pytest.fixture()
+@pytest.fixture(autouse=True)
 def use_local_magics_context(monkeypatch):
     if magics is not None:  # pragma: NO COVER
         local_context = magics.Context()
@@ -58,8 +58,7 @@ def use_local_magics_context(monkeypatch):
 @pytest.fixture(scope="session")
 def ipython():
     config = tools.default_config()
-    config.TerminalInteractiveShell.simple_prompt = True
-    shell = interactiveshell.TerminalInteractiveShell.instance(config=config)
+    shell = interactiveshell.InteractiveShell.instance(config=config)
     return shell
 
 
@@ -147,6 +146,8 @@ def test_context_with_default_credentials():
     """When Application Default Credentials are set, the context credentials
     will be created the first time it is called
     """
+    magics.context._credentials = None
+    magics.context._project = None
     assert magics.context._credentials is None
     assert magics.context._project is None
 
@@ -164,6 +165,16 @@ def test_context_with_default_credentials():
     assert default_mock.call_count == 2
 
 
+def test_context_fallback_when_bigquery_magics_none():
+    ctx = magics.Context()
+    credentials_mock = mock.create_autospec(
+        google.auth.credentials.Credentials, instance=True
+    )
+    with mock.patch("google.auth.default", return_value=(credentials_mock, "proj-123")):
+        assert ctx.credentials is credentials_mock
+        assert ctx.project == "proj-123"
+
+
 @pytest.mark.usefixtures("ipython_interactive")
 @pytest.mark.skipif(pandas is None, reason="Requires `pandas`")
 def test_context_with_default_connection(monkeypatch):
@@ -674,9 +685,11 @@ def test_bigquery_magic_with_bqstorage_from_argument(
         google.cloud.bigquery.job.QueryJob, instance=True
     )
     query_job_mock.to_dataframe.return_value = result
-    with run_query_patch as run_query_mock, (
-        bqstorage_client_patch
-    ), warnings.catch_warnings(record=True) as warned:
+    with (
+        run_query_patch as run_query_mock,
+        bqstorage_client_patch,
+        warnings.catch_warnings(record=True) as warned,
+    ):
         run_query_mock.return_value = query_job_mock
 
         return_value = ip.run_cell_magic("bigquery", "--use_bqstorage_api", sql)
@@ -842,11 +855,12 @@ def test_bigquery_magic_w_max_results_query_job_results_fails(monkeypatch):
     )
     query_job_mock.result.side_effect = [[], OSError]
 
-    with pytest.raises(
-        OSError
-    ), client_query_patch as client_query_mock, (
-        default_patch
-    ), close_transports_patch as close_transports:
+    with (
+        pytest.raises(OSError),
+        client_query_patch as client_query_mock,
+        default_patch,
+        close_transports_patch as close_transports,
+    ):
         client_query_mock.return_value = query_job_mock
         ip.run_cell_magic("bigquery", "--max_results=5", sql)
 
@@ -1965,9 +1979,10 @@ def test_bigquery_magic_nonexisting_query_variable(monkeypatch):
     ip.user_ns.pop("custom_query", None)  # Make sure the variable does NOT exist.
     cell_body = "$custom_query"  # Referring to a non-existing variable name.
 
-    with pytest.raises(
-        NameError, match=r".*custom_query does not exist.*"
-    ), run_query_patch as run_query_mock:
+    with (
+        pytest.raises(NameError, match=r".*custom_query does not exist.*"),
+        run_query_patch as run_query_mock,
+    ):
         ip.run_cell_magic("bigquery", "", cell_body)
 
     run_query_mock.assert_not_called()
@@ -1988,9 +2003,10 @@ def test_bigquery_magic_empty_query_variable_name(monkeypatch):
     )
     cell_body = "$"  # Not referring to any variable (name omitted).
 
-    with pytest.raises(
-        NameError, match=r"(?i).*missing query variable name.*"
-    ), run_query_patch as run_query_mock:
+    with (
+        pytest.raises(NameError, match=r"(?i).*missing query variable name.*"),
+        run_query_patch as run_query_mock,
+    ):
         ip.run_cell_magic("bigquery", "", cell_body)
 
     run_query_mock.assert_not_called()
@@ -2016,9 +2032,10 @@ def test_bigquery_magic_query_variable_non_string(ipython_ns_cleanup, monkeypatc
     ip.user_ns["custom_query"] = object()
     cell_body = "$custom_query"  # Referring to a non-string variable.
 
-    with pytest.raises(
-        TypeError, match=r".*must be a string or a bytes-like.*"
-    ), run_query_patch as run_query_mock:
+    with (
+        pytest.raises(TypeError, match=r".*must be a string or a bytes-like.*"),
+        run_query_patch as run_query_mock,
+    ):
         ip.run_cell_magic("bigquery", "", cell_body)
 
     run_query_mock.assert_not_called()
@@ -2183,9 +2200,11 @@ def test_bigquery_magic_create_dataset_fails(monkeypatch):
         autospec=True,
     )
 
-    with pytest.raises(
-        OSError
-    ), create_dataset_if_necessary_patch, close_transports_patch as close_transports:
+    with (
+        pytest.raises(OSError),
+        create_dataset_if_necessary_patch,
+        close_transports_patch as close_transports,
+    ):
         ip.run_cell_magic(
             "bigquery",
             "--destination_table dataset_id.table_id",
diff --git a/packages/google-cloud-bigquery/tests/unit/test_table.py b/packages/google-cloud-bigquery/tests/unit/test_table.py
index 5701143a62d4..135fbc725e47 100644
--- a/packages/google-cloud-bigquery/tests/unit/test_table.py
+++ b/packages/google-cloud-bigquery/tests/unit/test_table.py
@@ -22,18 +22,14 @@
 from unittest import mock
 import warnings
 
-import pytest
-
 import google.api_core.exceptions
+import pytest
 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 import _versions_helpers, exceptions, external_config, schema
+from google.cloud.bigquery.dataset import DatasetReference
 from google.cloud.bigquery.enums import DefaultPandasDTypes
 from google.cloud.bigquery.table import TableReference
-from google.cloud.bigquery.dataset import DatasetReference
 
 
 def _mock_client():
@@ -381,9 +377,7 @@ def test_from_api_repr(self):
     def test___repr__(self):
         dataset = DatasetReference("project1", "dataset1")
         table1 = self._make_one(dataset, "table1")
-        expected = (
-            "TableReference(DatasetReference('project1', 'dataset1'), " "'table1')"
-        )
+        expected = "TableReference(DatasetReference('project1', 'dataset1'), 'table1')"
         self.assertEqual(repr(table1), expected)
 
     def test___str__(self):
@@ -414,6 +408,7 @@ def _make_one(self, *args, **kw):
 
     def _setUpConstants(self):
         import datetime
+
         from google.cloud._helpers import UTC
 
         self.WHEN_TS = 1437767599.006
@@ -618,10 +613,11 @@ def test_ctor_string(self):
         self.assertEqual(table.table_id, "some_tbl")
 
     def test_ctor_tablelistitem(self):
-        from google.cloud.bigquery.table import Table, TableListItem
-
         import datetime
-        from google.cloud._helpers import _millis, UTC
+
+        from google.cloud._helpers import UTC, _millis
+
+        from google.cloud.bigquery.table import Table, TableListItem
 
         self.WHEN_TS = 1437767599.125
         self.EXP_TIME = datetime.datetime(2015, 8, 1, 23, 59, 59, tzinfo=UTC)
@@ -818,8 +814,8 @@ def test_schema_setter_valid_mapping_representation(self):
 
     def test_props_set_by_server(self):
         import datetime
-        from google.cloud._helpers import UTC
-        from google.cloud._helpers import _millis
+
+        from google.cloud._helpers import UTC, _millis
 
         CREATED = datetime.datetime(2015, 7, 29, 12, 13, 22, tzinfo=UTC)
         MODIFIED = datetime.datetime(2015, 7, 29, 14, 47, 15, tzinfo=UTC)
@@ -859,6 +855,7 @@ def test_snapshot_definition_not_set(self):
 
     def test_snapshot_definition_set(self):
         from google.cloud._helpers import UTC
+
         from google.cloud.bigquery.table import SnapshotDefinition
 
         dataset = DatasetReference(self.PROJECT, self.DS_ID)
@@ -893,6 +890,7 @@ def test_clone_definition_not_set(self):
 
     def test_clone_definition_set(self):
         from google.cloud._helpers import UTC
+
         from google.cloud.bigquery.table import CloneDefinition
 
         dataset = DatasetReference(self.PROJECT, self.DS_ID)
@@ -1162,6 +1160,7 @@ def test_expires_setter_bad_value(self):
 
     def test_expires_setter(self):
         import datetime
+
         from google.cloud._helpers import UTC
 
         WHEN = datetime.datetime(2015, 7, 28, 16, 39, tzinfo=UTC)
@@ -1374,8 +1373,8 @@ def test_from_api_repr_bare(self):
 
     def test_from_api_repr_w_properties(self):
         import datetime
-        from google.cloud._helpers import UTC
-        from google.cloud._helpers import _millis
+
+        from google.cloud._helpers import UTC, _millis
 
         RESOURCE = self._make_resource()
         RESOURCE["view"] = {"query": "select fullname, age from person_ages"}
@@ -1389,8 +1388,8 @@ def test_from_api_repr_w_properties(self):
 
     def test_from_api_repr_w_partial_streamingbuffer(self):
         import datetime
-        from google.cloud._helpers import UTC
-        from google.cloud._helpers import _millis
+
+        from google.cloud._helpers import UTC, _millis
 
         RESOURCE = self._make_resource()
         self.OLDEST_TIME = datetime.datetime(2015, 8, 1, 23, 59, 59, tzinfo=UTC)
@@ -1554,8 +1553,7 @@ def test__build_resource_w_custom_field_not_in__properties(self):
             table._build_resource(["bad"])
 
     def test_range_partitioning(self):
-        from google.cloud.bigquery.table import RangePartitioning
-        from google.cloud.bigquery.table import PartitionRange
+        from google.cloud.bigquery.table import PartitionRange, RangePartitioning
 
         table = self._make_one("proj.dset.tbl")
         assert table.range_partitioning is None
@@ -1588,8 +1586,7 @@ def test_require_partitioning_filter(self):
         assert table.require_partition_filter is None
 
     def test_time_partitioning_getter(self):
-        from google.cloud.bigquery.table import TimePartitioning
-        from google.cloud.bigquery.table import TimePartitioningType
+        from google.cloud.bigquery.table import TimePartitioning, TimePartitioningType
 
         dataset = DatasetReference(self.PROJECT, self.DS_ID)
         table_ref = dataset.table(self.TABLE_NAME)
@@ -1646,8 +1643,7 @@ def test_time_partitioning_getter_w_empty(self):
             self.assertIs(warning.category, PendingDeprecationWarning)
 
     def test_time_partitioning_setter(self):
-        from google.cloud.bigquery.table import TimePartitioning
-        from google.cloud.bigquery.table import TimePartitioningType
+        from google.cloud.bigquery.table import TimePartitioning, TimePartitioningType
 
         dataset = DatasetReference(self.PROJECT, self.DS_ID)
         table_ref = dataset.table(self.TABLE_NAME)
@@ -1837,9 +1833,7 @@ def test___repr__(self):
         dataset = DatasetReference("project1", "dataset1")
         table1 = self._make_one(TableReference(dataset, "table1"))
         expected = (
-            "Table(TableReference("
-            "DatasetReference('project1', 'dataset1'), "
-            "'table1'))"
+            "Table(TableReference(DatasetReference('project1', 'dataset1'), 'table1'))"
         )
         self.assertEqual(repr(table1), expected)
 
@@ -1903,7 +1897,7 @@ def _call_fut(self, mapping, schema):
         return _row_from_mapping(mapping, schema)
 
     def test__row_from_mapping_wo_schema(self):
-        from google.cloud.bigquery.table import Table, _TABLE_HAS_NO_SCHEMA
+        from google.cloud.bigquery.table import _TABLE_HAS_NO_SCHEMA, Table
 
         MAPPING = {"full_name": "Phred Phlyntstone", "age": 32}
         dataset = DatasetReference(self.PROJECT, self.DS_ID)
@@ -2241,6 +2235,7 @@ def test_ctor_empty_resource(self):
 
     def test_ctor_full_resource(self):
         from google.cloud._helpers import UTC
+
         from google.cloud.bigquery.table import TableReference
 
         resource = {
@@ -2372,6 +2367,7 @@ def test_ctor_empty_resource(self):
 
     def test_ctor_full_resource(self):
         from google.cloud._helpers import UTC
+
         from google.cloud.bigquery.table import TableReference
 
         resource = {
@@ -2581,8 +2577,7 @@ def _make_one_from_data(self, schema=(), rows=()):
         return self._make_one(_mock_client(), api_request, path, schema)
 
     def test_constructor(self):
-        from google.cloud.bigquery.table import _item_to_row
-        from google.cloud.bigquery.table import _rows_page_start
+        from google.cloud.bigquery.table import _item_to_row, _rows_page_start
 
         client = _mock_client()
         path = "/some/path"
@@ -2880,7 +2875,8 @@ def test__should_use_bqstorage_returns_true_if_no_cached_results(self):
 
     def test__should_use_bqstorage_returns_false_if_page_token_set(self):
         iterator = self._make_one(
-            page_token="abc", first_page_response=None  # not cached
+            page_token="abc",
+            first_page_response=None,  # not cached
         )
         result = iterator._should_use_bqstorage(
             bqstorage_client=None, create_bqstorage_client=True
@@ -2889,7 +2885,8 @@ def test__should_use_bqstorage_returns_false_if_page_token_set(self):
 
     def test__should_use_bqstorage_returns_false_if_max_results_set(self):
         iterator = self._make_one(
-            max_results=10, first_page_response=None  # not cached
+            max_results=10,
+            first_page_response=None,  # not cached
         )
         result = iterator._should_use_bqstorage(
             bqstorage_client=None, create_bqstorage_client=True
@@ -3047,12 +3044,12 @@ def test_to_arrow_iterable(self):
     def test_to_arrow_iterable_w_bqstorage(self):
         pyarrow = pytest.importorskip("pyarrow")
         pytest.importorskip("google.cloud.bigquery_storage")
-        from google.cloud import bigquery_storage
         from google.cloud.bigquery_storage_v1 import reader
         from google.cloud.bigquery_storage_v1.services.big_query_read.transports import (
             grpc as big_query_read_grpc_transport,
         )
 
+        from google.cloud import bigquery_storage
         from google.cloud.bigquery import schema
         from google.cloud.bigquery import table as mut
 
@@ -3217,6 +3214,7 @@ def test_to_arrow_w_nulls(self):
             "pyarrow", minversion=self.PYARROW_MINIMUM_VERSION
         )
         import pyarrow.types
+
         from google.cloud.bigquery.schema import SchemaField
 
         schema = [SchemaField("name", "STRING"), SchemaField("age", "INTEGER")]
@@ -3422,14 +3420,15 @@ def test_to_arrow_w_bqstorage(self):
         pytest.importorskip("numpy")
         pyarrow = pytest.importorskip("pyarrow")
         pytest.importorskip("google.cloud.bigquery_storage")
-        from google.cloud.bigquery import schema
-        from google.cloud.bigquery import table as mut
-        from google.cloud import bigquery_storage
         from google.cloud.bigquery_storage_v1 import reader
         from google.cloud.bigquery_storage_v1.services.big_query_read.transports import (
             grpc as big_query_read_grpc_transport,
         )
 
+        from google.cloud import bigquery_storage
+        from google.cloud.bigquery import schema
+        from google.cloud.bigquery import table as mut
+
         bqstorage_client = mock.create_autospec(bigquery_storage.BigQueryReadClient)
         bqstorage_client._transport = mock.create_autospec(
             big_query_read_grpc_transport.BigQueryReadGrpcTransport
@@ -3506,13 +3505,14 @@ def test_to_arrow_w_bqstorage_creates_client(self):
         pytest.importorskip("numpy")
         pytest.importorskip("pyarrow")
         pytest.importorskip("google.cloud.bigquery_storage")
-        from google.cloud.bigquery import schema
-        from google.cloud.bigquery import table as mut
-        from google.cloud import bigquery_storage
         from google.cloud.bigquery_storage_v1.services.big_query_read.transports import (
             grpc as big_query_read_grpc_transport,
         )
 
+        from google.cloud import bigquery_storage
+        from google.cloud.bigquery import schema
+        from google.cloud.bigquery import table as mut
+
         mock_client = _mock_client()
         bqstorage_client = mock.create_autospec(bigquery_storage.BigQueryReadClient)
         bqstorage_client._transport = mock.create_autospec(
@@ -3574,9 +3574,9 @@ def test_to_arrow_w_bqstorage_no_streams(self):
         pytest.importorskip("numpy")
         pyarrow = pytest.importorskip("pyarrow")
         pytest.importorskip("google.cloud.bigquery_storage")
+        from google.cloud import bigquery_storage
         from google.cloud.bigquery import schema
         from google.cloud.bigquery import table as mut
-        from google.cloud import bigquery_storage
 
         bqstorage_client = mock.create_autospec(bigquery_storage.BigQueryReadClient)
         session = bigquery_storage.types.ReadSession()
@@ -3749,14 +3749,15 @@ def test_to_dataframe_iterable_w_bqstorage(self):
         pandas = pytest.importorskip("pandas")
         pyarrow = pytest.importorskip("pyarrow")
         pytest.importorskip("google.cloud.bigquery_storage")
-        from google.cloud.bigquery import schema
-        from google.cloud.bigquery import table as mut
-        from google.cloud import bigquery_storage
         from google.cloud.bigquery_storage_v1 import reader
         from google.cloud.bigquery_storage_v1.services.big_query_read.transports import (
             grpc as big_query_read_grpc_transport,
         )
 
+        from google.cloud import bigquery_storage
+        from google.cloud.bigquery import schema
+        from google.cloud.bigquery import table as mut
+
         arrow_fields = [
             pyarrow.field("colA", pyarrow.int64()),
             # Not alphabetical to test column order.
@@ -3823,9 +3824,9 @@ def test_to_dataframe_iterable_w_bqstorage_max_results_warning(self):
         pytest.importorskip("numpy")
         pandas = pytest.importorskip("pandas")
         pytest.importorskip("google.cloud.bigquery_storage")
+        from google.cloud import bigquery_storage
         from google.cloud.bigquery import schema
         from google.cloud.bigquery import table as mut
-        from google.cloud import bigquery_storage
 
         bqstorage_client = mock.create_autospec(bigquery_storage.BigQueryReadClient)
 
@@ -4151,6 +4152,8 @@ def test_to_dataframe_tqdm_error(self):
                 # dependency and are unrelated to the code under test.
                 if "Pyparsing" in warning.category.__name__:
                     continue
+                if issubclass(warning.category, PendingDeprecationWarning):
+                    continue
                 self.assertIn(
                     warning.category,
                     [UserWarning, DeprecationWarning, tqdm.TqdmExperimentalWarning],
@@ -4176,6 +4179,7 @@ def test_to_dataframe_w_empty_results(self):
     def test_to_dataframe_w_various_types_nullable(self):
         pandas = pytest.importorskip("pandas")
         import datetime
+
         from google.cloud.bigquery.schema import SchemaField
 
         schema = [
@@ -4855,13 +4859,14 @@ def test_to_dataframe_w_bqstorage_creates_client(self):
         pytest.importorskip("numpy")
         pytest.importorskip("pandas")
         pytest.importorskip("google.cloud.bigquery_storage")
-        from google.cloud.bigquery import schema
-        from google.cloud.bigquery import table as mut
-        from google.cloud import bigquery_storage
         from google.cloud.bigquery_storage_v1.services.big_query_read.transports import (
             grpc as big_query_read_grpc_transport,
         )
 
+        from google.cloud import bigquery_storage
+        from google.cloud.bigquery import schema
+        from google.cloud.bigquery import table as mut
+
         mock_client = _mock_client()
         bqstorage_client = mock.create_autospec(bigquery_storage.BigQueryReadClient)
         bqstorage_client._transport = mock.create_autospec(
@@ -4889,9 +4894,9 @@ def test_to_dataframe_w_bqstorage_no_streams(self):
         pytest.importorskip("numpy")
         pytest.importorskip("pandas")
         pytest.importorskip("google.cloud.bigquery_storage")
+        from google.cloud import bigquery_storage
         from google.cloud.bigquery import schema
         from google.cloud.bigquery import table as mut
-        from google.cloud import bigquery_storage
 
         bqstorage_client = mock.create_autospec(bigquery_storage.BigQueryReadClient)
         session = bigquery_storage.types.ReadSession()
@@ -4919,8 +4924,8 @@ def test_to_dataframe_w_bqstorage_logs_session(self):
         pytest.importorskip("google.cloud.bigquery_storage")
         pytest.importorskip("pandas")
         pytest.importorskip("pyarrow")
-        from google.cloud.bigquery.table import Table
         from google.cloud import bigquery_storage
+        from google.cloud.bigquery.table import Table
 
         bqstorage_client = mock.create_autospec(bigquery_storage.BigQueryReadClient)
         session = bigquery_storage.types.ReadSession()
@@ -4944,10 +4949,11 @@ def test_to_dataframe_w_bqstorage_empty_streams(self):
         pytest.importorskip("google.cloud.bigquery_storage")
         pytest.importorskip("pandas")
         pyarrow = pytest.importorskip("pyarrow")
+        from google.cloud.bigquery_storage_v1 import reader
+
         from google.cloud import bigquery_storage
         from google.cloud.bigquery import schema
         from google.cloud.bigquery import table as mut
-        from google.cloud.bigquery_storage_v1 import reader
 
         arrow_fields = [
             pyarrow.field("colA", pyarrow.int64()),
@@ -4999,14 +5005,15 @@ def test_to_dataframe_w_bqstorage_nonempty(self):
         pytest.importorskip("google.cloud.bigquery_storage")
         pytest.importorskip("pandas")
         pyarrow = pytest.importorskip("pyarrow")
-        from google.cloud.bigquery import schema
-        from google.cloud.bigquery import table as mut
-        from google.cloud import bigquery_storage
         from google.cloud.bigquery_storage_v1 import reader
         from google.cloud.bigquery_storage_v1.services.big_query_read.transports import (
             grpc as big_query_read_grpc_transport,
         )
 
+        from google.cloud import bigquery_storage
+        from google.cloud.bigquery import schema
+        from google.cloud.bigquery import table as mut
+
         arrow_fields = [
             pyarrow.field("colA", pyarrow.int64()),
             # Not alphabetical to test column order.
@@ -5082,9 +5089,10 @@ def test_to_dataframe_w_bqstorage_multiple_streams_return_unique_index(self):
         bigquery_storage = pytest.importorskip("google.cloud.bigquery_storage")
         pytest.importorskip("pandas")
         pyarrow = pytest.importorskip("pyarrow")
+        from google.cloud.bigquery_storage_v1 import reader
+
         from google.cloud.bigquery import schema
         from google.cloud.bigquery import table as mut
-        from google.cloud.bigquery_storage_v1 import reader
 
         arrow_fields = [pyarrow.field("colA", pyarrow.int64())]
         arrow_schema = pyarrow.schema(arrow_fields)
@@ -5136,9 +5144,10 @@ def test_to_dataframe_w_bqstorage_updates_progress_bar(self):
         pytest.importorskip("pandas")
         pyarrow = pytest.importorskip("pyarrow")
         pytest.importorskip("tqdm")
+        from google.cloud.bigquery_storage_v1 import reader
+
         from google.cloud.bigquery import schema
         from google.cloud.bigquery import table as mut
-        from google.cloud.bigquery_storage_v1 import reader
 
         # Speed up testing.
         mut._PROGRESS_INTERVAL = 0.01
@@ -5213,9 +5222,10 @@ def test_to_dataframe_w_bqstorage_exits_on_keyboardinterrupt(self):
         bigquery_storage = pytest.importorskip("google.cloud.bigquery_storage")
         pytest.importorskip("pandas")
         pyarrow = pytest.importorskip("pyarrow")
+        from google.cloud.bigquery_storage_v1 import reader
+
         from google.cloud.bigquery import schema
         from google.cloud.bigquery import table as mut
-        from google.cloud.bigquery_storage_v1 import reader
 
         # Speed up testing.
         mut._PROGRESS_INTERVAL = 0.01
@@ -5389,14 +5399,15 @@ def test_to_dataframe_concat_categorical_dtype_w_pyarrow(self):
         pytest.importorskip("google.cloud.bigquery_storage")
         pandas = pytest.importorskip("pandas")
         pyarrow = pytest.importorskip("pyarrow")
-        from google.cloud import bigquery_storage
-        from google.cloud.bigquery import schema
-        from google.cloud.bigquery import table as mut
         from google.cloud.bigquery_storage_v1 import reader
         from google.cloud.bigquery_storage_v1.services.big_query_read.transports import (
             grpc as big_query_read_grpc_transport,
         )
 
+        from google.cloud import bigquery_storage
+        from google.cloud.bigquery import schema
+        from google.cloud.bigquery import table as mut
+
         arrow_fields = [
             # Not alphabetical to test column order.
             pyarrow.field("col_str", pyarrow.utf8()),
@@ -5604,8 +5615,7 @@ def test_to_geodataframe_no_geog(self):
         with self.assertRaisesRegex(
             TypeError,
             re.escape(
-                "There must be at least one GEOGRAPHY column"
-                " to create a GeoDataFrame"
+                "There must be at least one GEOGRAPHY column to create a GeoDataFrame"
             ),
         ):
             row_iterator.to_geodataframe(create_bqstorage_client=False)
@@ -5723,6 +5733,171 @@ def test_rowiterator_to_geodataframe_delegation(self, to_dataframe):
 
         self.assertEqual([v.__class__.__name__ for v in df.g], ["Point"])
 
+    def test_to_dataframe_delegated_updates_user_agent(self):
+        import sys
+
+        pytest.importorskip("db_dtypes")
+        pandas = pytest.importorskip("pandas")
+        mock_pandas_gbq = mock.Mock()
+        mock_pandas_gbq.__version__ = "1.0.0"
+
+        mock_client_info = mock.Mock()
+        mock_client_info.user_agent = "gl-python/3.10.0"
+
+        mock_client = _mock_client()
+        mock_client._connection = mock.Mock(_client_info=mock_client_info)
+
+        with mock.patch(
+            "google.cloud.bigquery._versions_helpers.PandasGBQVersions.is_delegation_supported",
+            new_callable=mock.PropertyMock,
+            return_value=True,
+        ):
+            with mock.patch(
+                "google.cloud.bigquery._versions_helpers.SUPPORTS_RANGE_PYARROW",
+                False,
+            ):
+                with mock.patch.dict(sys.modules, {"pandas_gbq": mock_pandas_gbq}):
+                    row_iterator = self._make_one_from_data(
+                        (("name", "STRING"),), (("foo",),)
+                    )
+                    row_iterator.client = mock_client
+                    df = row_iterator.to_dataframe(
+                        progress_bar_type="tqdm", timeout=5.0
+                    )
+                    self.assertIsInstance(df, pandas.DataFrame)
+                    self.assertEqual(
+                        mock_client_info.user_agent,
+                        "gl-python/3.10.0 pandas-gbq/1.0.0",
+                    )
+
+    def test_to_dataframe_delegated_does_not_duplicate_user_agent(self):
+        import sys
+
+        pytest.importorskip("db_dtypes")
+        pandas = pytest.importorskip("pandas")
+        mock_pandas_gbq = mock.Mock()
+        mock_pandas_gbq.__version__ = "1.0.0"
+
+        mock_client_info = mock.Mock()
+        mock_client_info.user_agent = "gl-python/3.10.0 pandas-gbq/1.0.0"
+
+        mock_client = _mock_client()
+        mock_client._connection = mock.Mock(_client_info=mock_client_info)
+
+        with mock.patch(
+            "google.cloud.bigquery._versions_helpers.PandasGBQVersions.is_delegation_supported",
+            new_callable=mock.PropertyMock,
+            return_value=True,
+        ):
+            with mock.patch(
+                "google.cloud.bigquery._versions_helpers.SUPPORTS_RANGE_PYARROW",
+                False,
+            ):
+                with mock.patch.dict(sys.modules, {"pandas_gbq": mock_pandas_gbq}):
+                    row_iterator = self._make_one_from_data(
+                        (("name", "STRING"),), (("foo",),)
+                    )
+                    row_iterator.client = mock_client
+                    df = row_iterator.to_dataframe(
+                        progress_bar_type="tqdm", timeout=5.0
+                    )
+                    self.assertIsInstance(df, pandas.DataFrame)
+                    self.assertEqual(
+                        mock_client_info.user_agent,
+                        "gl-python/3.10.0 pandas-gbq/1.0.0",
+                    )
+
+    def test_to_dataframe_delegated_when_client_info_is_none(self):
+        import sys
+
+        pytest.importorskip("db_dtypes")
+        pandas = pytest.importorskip("pandas")
+        mock_pandas_gbq = mock.Mock()
+        mock_pandas_gbq.__version__ = "1.0.0"
+
+        mock_client = _mock_client()
+        mock_client._connection = mock.Mock(_client_info=None)
+
+        with mock.patch(
+            "google.cloud.bigquery._versions_helpers.PandasGBQVersions.is_delegation_supported",
+            new_callable=mock.PropertyMock,
+            return_value=True,
+        ):
+            with mock.patch(
+                "google.cloud.bigquery._versions_helpers.SUPPORTS_RANGE_PYARROW",
+                False,
+            ):
+                with mock.patch.dict(sys.modules, {"pandas_gbq": mock_pandas_gbq}):
+                    row_iterator = self._make_one_from_data(
+                        (("name", "STRING"),), (("foo",),)
+                    )
+                    row_iterator.client = mock_client
+                    df = row_iterator.to_dataframe(
+                        progress_bar_type="tqdm", timeout=5.0
+                    )
+                    self.assertIsInstance(df, pandas.DataFrame)
+
+    def test_to_dataframe_delegated_when_user_agent_is_none(self):
+        import sys
+
+        pytest.importorskip("db_dtypes")
+        pandas = pytest.importorskip("pandas")
+        mock_pandas_gbq = mock.Mock()
+        mock_pandas_gbq.__version__ = "1.0.0"
+
+        mock_client_info = mock.Mock()
+        mock_client_info.user_agent = None
+
+        mock_client = _mock_client()
+        mock_client._connection = mock.Mock(_client_info=mock_client_info)
+
+        with mock.patch(
+            "google.cloud.bigquery._versions_helpers.PandasGBQVersions.is_delegation_supported",
+            new_callable=mock.PropertyMock,
+            return_value=True,
+        ):
+            with mock.patch(
+                "google.cloud.bigquery._versions_helpers.SUPPORTS_RANGE_PYARROW",
+                False,
+            ):
+                with mock.patch.dict(sys.modules, {"pandas_gbq": mock_pandas_gbq}):
+                    row_iterator = self._make_one_from_data(
+                        (("name", "STRING"),), (("foo",),)
+                    )
+                    row_iterator.client = mock_client
+                    df = row_iterator.to_dataframe(
+                        progress_bar_type="tqdm", timeout=5.0
+                    )
+                    self.assertIsInstance(df, pandas.DataFrame)
+                    self.assertEqual(
+                        mock_client_info.user_agent,
+                        "pandas-gbq/1.0.0",
+                    )
+
+    def test_to_geodataframe_updates_user_agent(self):
+        pytest.importorskip("geopandas")
+        row_iterator = self._make_one_from_data(
+            (("name", "STRING"), ("geog", "GEOGRAPHY")),
+            (("foo", "Point(0 0)"),),
+        )
+        mock_client_info = mock.Mock(user_agent="test-agent")
+        row_iterator.client._connection = mock.Mock(_client_info=mock_client_info)
+
+        with mock.patch(
+            "google.cloud.bigquery._versions_helpers.PandasGBQVersions.is_delegation_supported",
+            new_callable=mock.PropertyMock,
+            return_value=True,
+        ):
+            with mock.patch.object(row_iterator, "to_arrow") as mock_to_arrow:
+                import pyarrow as pa
+
+                mock_to_arrow.return_value = pa.RecordBatch.from_arrays(
+                    [pa.array(["foo"]), pa.array(["Point(0 0)"])],
+                    names=["name", "geog"],
+                )
+                _ = row_iterator.to_geodataframe(create_bqstorage_client=False)
+                self.assertIn("pandas-gbq/", mock_client_info.user_agent)
+
 
 class TestPartitionRange(unittest.TestCase):
     def _get_target_class(self):
@@ -6329,10 +6504,10 @@ def test_constructor_defaults(self):
 
     def test_constructor_explicit(self):
         from google.cloud.bigquery.table import (
-            PrimaryKey,
+            ColumnReference,
             ForeignKey,
+            PrimaryKey,
             TableReference,
-            ColumnReference,
         )
 
         primary_key = PrimaryKey(columns=["my_pk_id"])
@@ -6364,10 +6539,10 @@ def test_constructor_explicit_with_none(self):
 
     def test__eq__other_type(self):
         from google.cloud.bigquery.table import (
-            PrimaryKey,
+            ColumnReference,
             ForeignKey,
+            PrimaryKey,
             TableReference,
-            ColumnReference,
         )
 
         table_constraint = self._make_one(
@@ -6602,8 +6777,8 @@ def test_table_constraint_eq_parametrized(
         ColumnReference,
         ForeignKey,
         PrimaryKey,
-        TableReference,
         TableConstraints,
+        TableReference,
     )
 
     # Helper function to create a PrimaryKey object or None
@@ -6850,9 +7025,9 @@ def test_table_reference_to_bqstorage_v1_stable(table_path):
 def test_to_arrow_iterable_w_bqstorage_max_stream_count(preserve_order):
     pytest.importorskip("pandas")
     pytest.importorskip("google.cloud.bigquery_storage")
+    from google.cloud import bigquery_storage
     from google.cloud.bigquery import schema
     from google.cloud.bigquery import table as mut
-    from google.cloud import bigquery_storage
 
     bqstorage_client = mock.create_autospec(bigquery_storage.BigQueryReadClient)
     session = bigquery_storage.types.ReadSession()
@@ -6887,9 +7062,9 @@ def test_to_arrow_iterable_w_bqstorage_max_stream_count(preserve_order):
 def test_to_dataframe_iterable_w_bqstorage_max_stream_count(preserve_order):
     pytest.importorskip("pandas")
     pytest.importorskip("google.cloud.bigquery_storage")
+    from google.cloud import bigquery_storage
     from google.cloud.bigquery import schema
     from google.cloud.bigquery import table as mut
-    from google.cloud import bigquery_storage
 
     bqstorage_client = mock.create_autospec(bigquery_storage.BigQueryReadClient)
     session = bigquery_storage.types.ReadSession()