Merge remote-tracking branch 'github/main' into agg_callables

Author: TrevorBergeron

CHANGELOG.md

@@ -4,6 +4,21 @@
 
 [1]: https://pypi.org/project/bigframes/#history
 
+## [2.19.0](https://github.com/googleapis/python-bigquery-dataframes/compare/v2.18.0...v2.19.0) (2025-09-09)
+
+
+### Features
+
+* Add str.join method ([#2054](https://github.com/googleapis/python-bigquery-dataframes/issues/2054)) ([8804ada](https://github.com/googleapis/python-bigquery-dataframes/commit/8804adaf8ba23fdcad6e42a7bf034bd0a11c890f))
+* Support display.max_colwidth option ([#2053](https://github.com/googleapis/python-bigquery-dataframes/issues/2053)) ([5229e07](https://github.com/googleapis/python-bigquery-dataframes/commit/5229e07b4535c01b0cdbd731455ff225a373b5c8))
+* Support VPC egress setting in remote function ([#2059](https://github.com/googleapis/python-bigquery-dataframes/issues/2059)) ([5df779d](https://github.com/googleapis/python-bigquery-dataframes/commit/5df779d4f421d3ba777cfd928d99ca2e8a3f79ad))
+
+
+### Bug Fixes
+
+* Fix issue mishandling chunked array while loading data ([#2051](https://github.com/googleapis/python-bigquery-dataframes/issues/2051)) ([873d0ee](https://github.com/googleapis/python-bigquery-dataframes/commit/873d0eee474ed34f1d5164c37383f2737dbec4db))
+* Remove warning for slot_millis_sum ([#2047](https://github.com/googleapis/python-bigquery-dataframes/issues/2047)) ([425a691](https://github.com/googleapis/python-bigquery-dataframes/commit/425a6917d5442eeb4df486c6eed1fd136bbcedfb))
+
 ## [2.18.0](https://github.com/googleapis/python-bigquery-dataframes/compare/v2.17.0...v2.18.0) (2025-09-03)
 
 

bigframes/_config/auth.py

@@ -0,0 +1,57 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import threading
+from typing import Optional
+
+import google.auth.credentials
+import google.auth.transport.requests
+import pydata_google_auth
+
+_SCOPES = ["https://www.googleapis.com/auth/cloud-platform"]
+
+# Put the lock here rather than in BigQueryOptions so that BigQueryOptions
+# remains deepcopy-able.
+_AUTH_LOCK = threading.Lock()
+_cached_credentials: Optional[google.auth.credentials.Credentials] = None
+_cached_project_default: Optional[str] = None
+
+
+def get_default_credentials_with_project() -> tuple[
+    google.auth.credentials.Credentials, Optional[str]
+]:
+    global _AUTH_LOCK, _cached_credentials, _cached_project_default
+
+    with _AUTH_LOCK:
+        if _cached_credentials is not None:
+            return _cached_credentials, _cached_project_default
+
+        _cached_credentials, _cached_project_default = pydata_google_auth.default(
+            scopes=_SCOPES, use_local_webserver=False
+        )
+
+        # Ensure an access token is available.
+        _cached_credentials.refresh(google.auth.transport.requests.Request())
+
+    return _cached_credentials, _cached_project_default
+
+
+def reset_default_credentials_and_project():
+    global _AUTH_LOCK, _cached_credentials, _cached_project_default
+
+    with _AUTH_LOCK:
+        _cached_credentials = None
+        _cached_project_default = None

bigframes/_config/bigquery_options.py

@@ -22,6 +22,7 @@
 import google.auth.credentials
 import requests.adapters
 
+import bigframes._config.auth
 import bigframes._importing
 import bigframes.enums
 import bigframes.exceptions as bfe
@@ -37,6 +38,7 @@
 
 def _get_validated_location(value: Optional[str]) -> Optional[str]:
     import bigframes._tools.strings
+    import bigframes.constants
 
     if value is None or value in bigframes.constants.ALL_BIGQUERY_LOCATIONS:
         return value
@@ -141,20 +143,52 @@ def application_name(self, value: Optional[str]):
             )
         self._application_name = value
 
+    def _try_set_default_credentials_and_project(
+        self,
+    ) -> tuple[google.auth.credentials.Credentials, Optional[str]]:
+        # Don't fetch credentials or project if credentials is already set.
+        # If it's set, we've already authenticated, so if the user wants to
+        # re-auth, they should explicitly reset the credentials.
+        if self._credentials is not None:
+            return self._credentials, self._project
+
+        (
+            credentials,
+            credentials_project,
+        ) = bigframes._config.auth.get_default_credentials_with_project()
+        self._credentials = credentials
+
+        # Avoid overriding an explicitly set project with a default value.
+        if self._project is None:
+            self._project = credentials_project
+
+        return credentials, self._project
+
     @property
-    def credentials(self) -> Optional[google.auth.credentials.Credentials]:
+    def credentials(self) -> google.auth.credentials.Credentials:
         """The OAuth2 credentials to use for this client.
 
+        Set to None to force re-authentication.
+
         Returns:
             None or google.auth.credentials.Credentials:
                 google.auth.credentials.Credentials if exists; otherwise None.
         """
-        return self._credentials
+        if self._credentials:
+            return self._credentials
+
+        credentials, _ = self._try_set_default_credentials_and_project()
+        return credentials
 
     @credentials.setter
     def credentials(self, value: Optional[google.auth.credentials.Credentials]):
         if self._session_started and self._credentials is not value:
             raise ValueError(SESSION_STARTED_MESSAGE.format(attribute="credentials"))
+
+        if value is None:
+            # The user has _explicitly_ asked that we re-authenticate.
+            bigframes._config.auth.reset_default_credentials_and_project()
+
         self._credentials = value
 
     @property
@@ -183,7 +217,11 @@ def project(self) -> Optional[str]:
             None or str:
                 Google Cloud project ID as a string; otherwise None.
         """
-        return self._project
+        if self._project:
+            return self._project
+
+        _, project = self._try_set_default_credentials_and_project()
+        return project
 
     @project.setter
     def project(self, value: Optional[str]):

bigframes/core/compile/ibis_compiler/aggregate_compiler.py

@@ -676,6 +676,29 @@ def _(
     ).to_expr()
 
 
+@compile_ordered_unary_agg.register
+def _(
+    op: agg_ops.StringAggOp,
+    column: ibis_types.Column,
+    window=None,
+    order_by: typing.Sequence[ibis_types.Value] = [],
+) -> ibis_types.ArrayValue:
+    if window is not None:
+        raise NotImplementedError(
+            f"StringAgg with windowing is not supported. {constants.FEEDBACK_LINK}"
+        )
+
+    return (
+        ibis_ops.StringAgg(
+            column,  # type: ignore
+            sep=op.sep,  # type: ignore
+            order_by=order_by,  # type: ignore
+        )
+        .to_expr()
+        .fill_null(ibis_types.literal(""))
+    )
+
+
 @compile_binary_agg.register
 def _(
     op: agg_ops.CorrOp, left: ibis_types.Column, right: ibis_types.Column, window=None

bigframes/core/compile/ibis_compiler/scalar_op_registry.py

@@ -1216,11 +1216,18 @@ def to_arry_op_impl(*values: ibis_types.Value):
 def array_reduce_op_impl(x: ibis_types.Value, op: ops.ArrayReduceOp):
     import bigframes.core.compile.ibis_compiler.aggregate_compiler as agg_compilers
 
-    return typing.cast(ibis_types.ArrayValue, x).reduce(
-        lambda arr_vals: agg_compilers.compile_unary_agg(
-            op.aggregation, typing.cast(ibis_types.Column, arr_vals)
+    if op.aggregation.order_independent:
+        return typing.cast(ibis_types.ArrayValue, x).reduce(
+            lambda arr_vals: agg_compilers.compile_unary_agg(
+                op.aggregation, typing.cast(ibis_types.Column, arr_vals)
+            )
+        )
+    else:
+        return typing.cast(ibis_types.ArrayValue, x).reduce(
+            lambda arr_vals: agg_compilers.compile_ordered_unary_agg(
+                op.aggregation, typing.cast(ibis_types.Column, arr_vals)
+            )
         )
-    )
 
 
 # JSON Ops

bigframes/functions/_function_client.py

@@ -51,6 +51,15 @@
     }
 )
 
+# https://cloud.google.com/functions/docs/reference/rest/v2/projects.locations.functions#vpconnectoregresssettings
+_VPC_EGRESS_SETTINGS_MAP = types.MappingProxyType(
+    {
+        "all": functions_v2.ServiceConfig.VpcConnectorEgressSettings.ALL_TRAFFIC,
+        "private-ranges-only": functions_v2.ServiceConfig.VpcConnectorEgressSettings.PRIVATE_RANGES_ONLY,
+        "unspecified": functions_v2.ServiceConfig.VpcConnectorEgressSettings.VPC_CONNECTOR_EGRESS_SETTINGS_UNSPECIFIED,
+    }
+)
+
 # BQ managed functions (@udf) currently only support Python 3.11.
 _MANAGED_FUNC_PYTHON_VERSION = "python-3.11"
 
@@ -375,6 +384,7 @@ def create_cloud_function(
         max_instance_count=None,
         is_row_processor=False,
         vpc_connector=None,
+        vpc_connector_egress_settings="private-ranges-only",
         memory_mib=1024,
         ingress_settings="internal-only",
     ):
@@ -472,6 +482,15 @@ def create_cloud_function(
                 function.service_config.max_instance_count = max_instance_count
             if vpc_connector is not None:
                 function.service_config.vpc_connector = vpc_connector
+                if vpc_connector_egress_settings not in _VPC_EGRESS_SETTINGS_MAP:
+                    raise bf_formatting.create_exception_with_feedback_link(
+                        ValueError,
+                        f"'{vpc_connector_egress_settings}' not one of the supported vpc egress settings values: {list(_VPC_EGRESS_SETTINGS_MAP)}",
+                    )
+                function.service_config.vpc_connector_egress_settings = cast(
+                    functions_v2.ServiceConfig.VpcConnectorEgressSettings,
+                    _VPC_EGRESS_SETTINGS_MAP[vpc_connector_egress_settings],
+                )
             function.service_config.service_account_email = (
                 self._cloud_function_service_account
             )
@@ -532,6 +551,7 @@ def provision_bq_remote_function(
         cloud_function_max_instance_count,
         is_row_processor,
         cloud_function_vpc_connector,
+        cloud_function_vpc_connector_egress_settings,
         cloud_function_memory_mib,
         cloud_function_ingress_settings,
         bq_metadata,
@@ -580,6 +600,7 @@ def provision_bq_remote_function(
                 max_instance_count=cloud_function_max_instance_count,
                 is_row_processor=is_row_processor,
                 vpc_connector=cloud_function_vpc_connector,
+                vpc_connector_egress_settings=cloud_function_vpc_connector_egress_settings,
                 memory_mib=cloud_function_memory_mib,
                 ingress_settings=cloud_function_ingress_settings,
             )

bigframes/functions/_function_session.py

@@ -245,6 +245,9 @@ def remote_function(
         cloud_function_timeout: Optional[int] = 600,
         cloud_function_max_instances: Optional[int] = None,
         cloud_function_vpc_connector: Optional[str] = None,
+        cloud_function_vpc_connector_egress_settings: Literal[
+            "all", "private-ranges-only", "unspecified"
+        ] = "private-ranges-only",
         cloud_function_memory_mib: Optional[int] = 1024,
         cloud_function_ingress_settings: Literal[
             "all", "internal-only", "internal-and-gclb"
@@ -425,6 +428,13 @@ def remote_function(
                 function. This is useful if your code needs access to data or
                 service(s) that are on a VPC network. See for more details
                 https://cloud.google.com/functions/docs/networking/connecting-vpc.
+            cloud_function_vpc_connector_egress_settings (str, Optional):
+                Egress settings for the VPC connector, controlling what outbound
+                traffic is routed through the VPC connector.
+                Options are: `all`, `private-ranges-only`, or `unspecified`.
+                If not specified, `private-ranges-only` is used by default.
+                See for more details
+                https://cloud.google.com/run/docs/configuring/vpc-connectors#egress-job.
             cloud_function_memory_mib (int, Optional):
                 The amounts of memory (in mebibytes) to allocate for the cloud
                 function (2nd gen) created. This also dictates a corresponding
@@ -616,6 +626,7 @@ def wrapper(func):
                 cloud_function_max_instance_count=cloud_function_max_instances,
                 is_row_processor=is_row_processor,
                 cloud_function_vpc_connector=cloud_function_vpc_connector,
+                cloud_function_vpc_connector_egress_settings=cloud_function_vpc_connector_egress_settings,
                 cloud_function_memory_mib=cloud_function_memory_mib,
                 cloud_function_ingress_settings=cloud_function_ingress_settings,
                 bq_metadata=bqrf_metadata,

bigframes/operations/aggregations.py

@@ -380,9 +380,26 @@ def skips_nulls(self):
         return True
 
     def output_type(self, *input_types: dtypes.ExpressionType) -> dtypes.ExpressionType:
-        return pd.ArrowDtype(
-            pa.list_(dtypes.bigframes_dtype_to_arrow_dtype(input_types[0]))
-        )
+        return dtypes.list_type(input_types[0])
+
+
+@dataclasses.dataclass(frozen=True)
+class StringAggOp(UnaryAggregateOp):
+    name: ClassVar[str] = "string_agg"
+    sep: str = ","
+
+    @property
+    def order_independent(self):
+        return False
+
+    @property
+    def skips_nulls(self):
+        return True
+
+    def output_type(self, *input_types: dtypes.ExpressionType) -> dtypes.ExpressionType:
+        if input_types[0] != dtypes.STRING_DTYPE:
+            raise TypeError(f"Type {input_types[0]} is not string-like")
+        return dtypes.STRING_DTYPE
 
 
 @dataclasses.dataclass(frozen=True)

bigframes/operations/strings.py

@@ -24,6 +24,7 @@
 import bigframes.dataframe as df
 import bigframes.operations as ops
 from bigframes.operations._op_converters import convert_index, convert_slice
+import bigframes.operations.aggregations as agg_ops
 import bigframes.operations.base
 import bigframes.series as series
 
@@ -295,6 +296,11 @@ def cat(
     ) -> series.Series:
         return self._apply_binary_op(others, ops.strconcat_op, alignment=join)
 
+    def join(self, sep: str) -> series.Series:
+        return self._apply_unary_op(
+            ops.ArrayReduceOp(aggregation=agg_ops.StringAggOp(sep=sep))
+        )
+
     def to_blob(self, connection: Optional[str] = None) -> series.Series:
         """Create a BigFrames Blob series from a series of URIs.
 

bigframes/pandas/__init__.py

@@ -87,6 +87,9 @@ def remote_function(
     cloud_function_timeout: Optional[int] = 600,
     cloud_function_max_instances: Optional[int] = None,
     cloud_function_vpc_connector: Optional[str] = None,
+    cloud_function_vpc_connector_egress_settings: Literal[
+        "all", "private-ranges-only", "unspecified"
+    ] = "private-ranges-only",
     cloud_function_memory_mib: Optional[int] = 1024,
     cloud_function_ingress_settings: Literal[
         "all", "internal-only", "internal-and-gclb"
@@ -109,6 +112,7 @@ def remote_function(
         cloud_function_timeout=cloud_function_timeout,
         cloud_function_max_instances=cloud_function_max_instances,
         cloud_function_vpc_connector=cloud_function_vpc_connector,
+        cloud_function_vpc_connector_egress_settings=cloud_function_vpc_connector_egress_settings,
         cloud_function_memory_mib=cloud_function_memory_mib,
         cloud_function_ingress_settings=cloud_function_ingress_settings,
         cloud_build_service_account=cloud_build_service_account,