feat: implement pandas-gbq delegation check and telemetry

Author: shuoweil

packages/google-cloud-bigquery/google/cloud/bigquery/_versions_helpers.py

@@ -16,10 +16,8 @@
 from typing import Any
 
 import packaging.version
-
 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 +245,49 @@ 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 None:
+            try:
+                import pandas_gbq  # type: ignore
+
+                self._installed_version = packaging.version.parse(
+                    getattr(pandas_gbq, "__version__", "0.0.0")
+                )
+            except ImportError:
+                self._installed_version = packaging.version.parse("0.0.0")
+
+        return self._installed_version
+
+    @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 None:
+            try:
+                import pandas_gbq  # type: ignore
+
+                self._delegation_api_version = getattr(
+                    pandas_gbq, "_internal_delegation_api_version", 0
+                )
+            except ImportError:
+                self._delegation_api_version = 0
+
+        return self._delegation_api_version
+
+    @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()

packages/google-cloud-bigquery/google/cloud/bigquery/table.py

@@ -21,9 +21,8 @@
 import functools
 import operator
 import typing
-from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union, Sequence
-
 import warnings
+from typing import Any, Dict, Iterable, Iterator, List, Optional, Sequence, Tuple, Union
 
 try:
     import pandas  # type: ignore
@@ -56,30 +55,33 @@
     _read_wkt = wkt.loads
 
 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.api_core.page_iterator import HTTPIterator
+from google.cloud.bigquery import (
+    _helpers,
+    _pandas_helpers,
+    _string_references,
+    _versions_helpers,
+    external_config,
+)
 from google.cloud.bigquery import exceptions as bq_exceptions
+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
 
@@ -2709,6 +2711,14 @@ def to_dataframe(
                 is not supported dtype.
 
         """
+        if not _versions_helpers.PANDAS_GBQ_VERSIONS.is_delegation_supported:
+            warnings.warn(
+                "Retrieving dataframes via the core client is deprecated. "
+                "Please install 'pandas-gbq' for the new high-performance backend.",
+                PendingDeprecationWarning,
+                stacklevel=2,
+            )
+
         _pandas_helpers.verify_pandas_imports()
 
         if geography_as_object and shapely is None:
@@ -2801,6 +2811,37 @@ def to_dataframe(
             create_bqstorage_client = False
             bqstorage_client = None
 
+        if _versions_helpers.PANDAS_GBQ_VERSIONS.is_delegation_supported:
+            import pandas_gbq  # type: ignore
+
+            if self.client and hasattr(self.client, "_connection") and hasattr(self.client._connection, "_client_info"):
+                client_info = self.client._connection._client_info
+                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()
+
+            return pandas_gbq.pandas.from_row_iterator(
+                self,
+                bqstorage_client=bqstorage_client,
+                dtypes=dtypes,
+                progress_bar_type=progress_bar_type,
+                create_bqstorage_client=create_bqstorage_client,
+                geography_as_object=geography_as_object,
+                bool_dtype=bool_dtype,
+                int_dtype=int_dtype,
+                float_dtype=float_dtype,
+                string_dtype=string_dtype,
+                date_dtype=date_dtype,
+                datetime_dtype=datetime_dtype,
+                time_dtype=time_dtype,
+                timestamp_dtype=timestamp_dtype,
+                range_date_dtype=range_date_dtype,
+                range_datetime_dtype=range_datetime_dtype,
+                range_timestamp_dtype=range_timestamp_dtype,
+                timeout=timeout,
+            )
+
         record_batch = self.to_arrow(
             progress_bar_type=progress_bar_type,
             bqstorage_client=bqstorage_client,