feat: create deploy_remote_function and deploy_udf functions to immediately deploy functions to BigQuery (#1832)

* Refactor function deployment to avoid code duplication

This commit refactors the implementation of immediate deployment
for remote functions and UDFs to eliminate code duplication introduced
in a previous commit.

Changes:
- The `remote_function` and `udf` methods in
  `bigframes.functions._function_session.FunctionSession` now accept
  an optional `deploy_immediately: bool` parameter (defaulting to `False`).
  The previous `deploy_remote_function` and `deploy_udf` methods in
  `FunctionSession` have been removed, and their logic is now
  incorporated into the unified methods.
- The public API functions `bigframes.pandas.deploy_remote_function`
  and `bigframes.pandas.deploy_udf` now call the corresponding
  `FunctionSession` methods with `deploy_immediately=True`.
- The public API functions `bigframes.pandas.remote_function` and
  `bigframes.pandas.udf` call the `FunctionSession` methods
  with `deploy_immediately=False` (relying on the default).
- Unit tests in `tests/unit/functions/test_remote_function.py` have
  been updated to patch the unified `FunctionSession` methods and
  verify the correct `deploy_immediately` boolean is passed based on
  which public API function is called.

Note: The underlying provisioning logic in `FunctionSession` currently
deploys functions immediately regardless of the `deploy_immediately`
flag. This flag serves as an indicator of intent and allows for
future enhancements to support true lazy deployment if desired, without
further API changes.

* Refactor function deployment to use distinct methods

This commit corrects a previous refactoring attempt to eliminate
code duplication and properly separates immediate-deployment functions
from standard (potentially lazy) functions.

Changes:
- `bigframes.functions._function_session.FunctionSession` now has
  distinct methods: `remote_function`, `udf`,
  `deploy_remote_function`, and `deploy_udf`. The
  `deploy_immediately` flag has been removed from this class.
- `deploy_remote_function` and `deploy_udf` methods in
  `FunctionSession` are responsible for ensuring immediate
  deployment by calling the underlying provisioning logic directly.
  The standard `remote_function` and `udf` methods in
  `FunctionSession` also currently call this provisioning logic,
  meaning all functions are deployed immediately as of now, but the
  structure allows for future lazy evaluation for standard functions
  without changing the deploy variants' contract.
- Public API functions in `bigframes.pandas` (`remote_function`,
  `udf`, `deploy_remote_function`, `deploy_udf`) now correctly
  delegate to their corresponding distinct methods in `FunctionSession`
  (via the `Session` object).
- Unit tests in `tests/unit/functions/test_remote_function.py` have
  been updated to mock and verify calls to the correct distinct
  methods on `bigframes.session.Session`.

This resolves the issue of using a boolean flag to control
deployment type and instead relies on calling specific, dedicated
methods for immediate deployment, aligning with your request.

* Simplify internal deploy_remote_function and deploy_udf calls

This commit simplifies the implementation of
`deploy_remote_function` and `deploy_udf` within
`bigframes.functions._function_session.FunctionSession`.

Given that the standard `remote_function` and `udf` methods in
`FunctionSession` already perform immediate deployment of resources
(as the underlying provisioning logic they call is immediate),
the `deploy_remote_function` and `deploy_udf` methods in the
same class are simplified to directly call `self.remote_function(...)`
and `self.udf(...)` respectively.

This change makes the distinction between the `deploy_` variants and
the standard variants in `FunctionSession` primarily a matter of
semantic clarity and intent at that level; both paths currently
result in immediate deployment. The public API in `bigframes.pandas`
continues to offer distinct `deploy_` functions that call these
`FunctionSession.deploy_` methods, preserving your user-facing API
and its documented behavior of immediate deployment.

No changes were needed for the public API in `bigframes.pandas` or
the unit tests, as they were already aligned with calling distinct
methods on the `Session` object, which in turn calls the
now-simplified `FunctionSession` methods.

* add tests and use kwargs

* add missing func argument to bpd

---------

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: tswast

bigframes/core/global_session.py

@@ -110,8 +110,8 @@ def get_global_session():
 _T = TypeVar("_T")
 
 
-def with_default_session(func: Callable[..., _T], *args, **kwargs) -> _T:
-    return func(get_global_session(), *args, **kwargs)
+def with_default_session(func_: Callable[..., _T], *args, **kwargs) -> _T:
+    return func_(get_global_session(), *args, **kwargs)
 
 
 class _GlobalSessionContext:

bigframes/functions/_function_session.py

@@ -668,6 +668,30 @@ def wrapper(func):
 
         return wrapper
 
+    def deploy_remote_function(
+        self,
+        func,
+        **kwargs,
+    ):
+        """Orchestrates the creation of a BigQuery remote function that deploys immediately.
+
+        This method ensures that the remote function is created and available for
+        use in BigQuery as soon as this call is made.
+
+        Args:
+            kwargs:
+                All arguments are passed directly to
+                :meth:`~bigframes.session.Session.remote_function`.  Please see
+                its docstring for parameter details.
+
+        Returns:
+            A wrapped remote function, usable in
+            :meth:`~bigframes.series.Series.apply`.
+        """
+        # TODO(tswast): If we update remote_function to defer deployment, update
+        # this method to deploy immediately.
+        return self.remote_function(**kwargs)(func)
+
     def udf(
         self,
         input_types: Union[None, type, Sequence[type]] = None,
@@ -866,6 +890,32 @@ def wrapper(func):
 
         return wrapper
 
+    def deploy_udf(
+        self,
+        func,
+        **kwargs,
+    ):
+        """Orchestrates the creation of a BigQuery UDF that deploys immediately.
+
+        This method ensures that the UDF is created and available for
+        use in BigQuery as soon as this call is made.
+
+        Args:
+            func:
+                Function to deploy.
+            kwargs:
+                All arguments are passed directly to
+                :meth:`~bigframes.session.Session.udf`.  Please see
+                its docstring for parameter details.
+
+        Returns:
+            A wrapped Python user defined function, usable in
+            :meth:`~bigframes.series.Series.apply`.
+        """
+        # TODO(tswast): If we update udf to defer deployment, update this method
+        # to deploy immediately.
+        return self.udf(**kwargs)(func)
+
 
 def _convert_row_processor_sig(
     signature: inspect.Signature,

bigframes/pandas/__init__.py

@@ -117,6 +117,22 @@ def remote_function(
 remote_function.__doc__ = inspect.getdoc(bigframes.session.Session.remote_function)
 
 
+def deploy_remote_function(
+    func,
+    **kwargs,
+):
+    return global_session.with_default_session(
+        bigframes.session.Session.deploy_remote_function,
+        func=func,
+        **kwargs,
+    )
+
+
+deploy_remote_function.__doc__ = inspect.getdoc(
+    bigframes.session.Session.deploy_remote_function
+)
+
+
 def udf(
     *,
     input_types: Union[None, type, Sequence[type]] = None,
@@ -140,6 +156,20 @@ def udf(
 udf.__doc__ = inspect.getdoc(bigframes.session.Session.udf)
 
 
+def deploy_udf(
+    func,
+    **kwargs,
+):
+    return global_session.with_default_session(
+        bigframes.session.Session.deploy_udf,
+        func=func,
+        **kwargs,
+    )
+
+
+deploy_udf.__doc__ = inspect.getdoc(bigframes.session.Session.deploy_udf)
+
+
 @typing.overload
 def to_datetime(
     arg: Union[
@@ -330,6 +360,8 @@ def reset_session():
     clean_up_by_session_id,
     concat,
     cut,
+    deploy_remote_function,
+    deploy_udf,
     get_default_session_id,
     get_dummies,
     merge,

bigframes/session/__init__.py

@@ -1343,6 +1343,40 @@ def _check_file_size(self, filepath: str):
                 "for large files to avoid loading the file into local memory."
             )
 
+    def deploy_remote_function(
+        self,
+        func,
+        **kwargs,
+    ):
+        """Orchestrates the creation of a BigQuery remote function that deploys immediately.
+
+        This method ensures that the remote function is created and available for
+        use in BigQuery as soon as this call is made.
+
+        Args:
+            func:
+                Function to deploy.
+            kwargs:
+                All arguments are passed directly to
+                :meth:`~bigframes.session.Session.remote_function`.  Please see
+                its docstring for parameter details.
+
+        Returns:
+            A wrapped remote function, usable in
+            :meth:`~bigframes.series.Series.apply`.
+        """
+        return self._function_session.deploy_remote_function(
+            func,
+            # Session-provided arguments.
+            session=self,
+            bigquery_client=self._clients_provider.bqclient,
+            bigquery_connection_client=self._clients_provider.bqconnectionclient,
+            cloud_functions_client=self._clients_provider.cloudfunctionsclient,
+            resource_manager_client=self._clients_provider.resourcemanagerclient,
+            # User-provided arguments.
+            **kwargs,
+        )
+
     def remote_function(
         self,
         # Make sure that the input/output types, and dataset can be used
@@ -1565,9 +1599,15 @@ def remote_function(
                 `bigframes_remote_function` - The bigquery remote function capable of calling into `bigframes_cloud_function`.
         """
         return self._function_session.remote_function(
+            # Session-provided arguments.
+            session=self,
+            bigquery_client=self._clients_provider.bqclient,
+            bigquery_connection_client=self._clients_provider.bqconnectionclient,
+            cloud_functions_client=self._clients_provider.cloudfunctionsclient,
+            resource_manager_client=self._clients_provider.resourcemanagerclient,
+            # User-provided arguments.
             input_types=input_types,
             output_type=output_type,
-            session=self,
             dataset=dataset,
             bigquery_connection=bigquery_connection,
             reuse=reuse,
@@ -1585,6 +1625,37 @@ def remote_function(
             cloud_build_service_account=cloud_build_service_account,
         )
 
+    def deploy_udf(
+        self,
+        func,
+        **kwargs,
+    ):
+        """Orchestrates the creation of a BigQuery UDF that deploys immediately.
+
+        This method ensures that the UDF is created and available for
+        use in BigQuery as soon as this call is made.
+
+        Args:
+            func:
+                Function to deploy.
+            kwargs:
+                All arguments are passed directly to
+                :meth:`~bigframes.session.Session.udf`.  Please see
+                its docstring for parameter details.
+
+        Returns:
+            A wrapped Python user defined function, usable in
+            :meth:`~bigframes.series.Series.apply`.
+        """
+        return self._function_session.deploy_udf(
+            func,
+            # Session-provided arguments.
+            session=self,
+            bigquery_client=self._clients_provider.bqclient,
+            # User-provided arguments.
+            **kwargs,
+        )
+
     def udf(
         self,
         *,
@@ -1726,9 +1797,12 @@ def udf(
                 deployed for the user defined code.
         """
         return self._function_session.udf(
+            # Session-provided arguments.
+            session=self,
+            bigquery_client=self._clients_provider.bqclient,
+            # User-provided arguments.
             input_types=input_types,
             output_type=output_type,
-            session=self,
             dataset=dataset,
             bigquery_connection=bigquery_connection,
             name=name,

tests/unit/functions/test_remote_function.py

@@ -89,3 +89,57 @@ def function_without_return_annotation(myparam: int):
         match="'output_type' was not set .* missing a return type annotation",
     ):
         remote_function_decorator(function_without_return_annotation)
+
+
+def test_deploy_remote_function():
+    session = mocks.create_bigquery_session()
+
+    def my_remote_func(x: int) -> int:
+        return x * 2
+
+    deployed = session.deploy_remote_function(
+        my_remote_func, cloud_function_service_account="test_sa@example.com"
+    )
+
+    # Test that the function would have been deployed somewhere.
+    assert deployed.bigframes_bigquery_function
+
+
+def test_deploy_remote_function_with_name():
+    session = mocks.create_bigquery_session()
+
+    def my_remote_func(x: int) -> int:
+        return x * 2
+
+    deployed = session.deploy_remote_function(
+        my_remote_func,
+        name="my_custom_name",
+        cloud_function_service_account="test_sa@example.com",
+    )
+
+    # Test that the function would have been deployed somewhere.
+    assert "my_custom_name" in deployed.bigframes_bigquery_function
+
+
+def test_deploy_udf():
+    session = mocks.create_bigquery_session()
+
+    def my_remote_func(x: int) -> int:
+        return x * 2
+
+    deployed = session.deploy_udf(my_remote_func)
+
+    # Test that the function would have been deployed somewhere.
+    assert deployed.bigframes_bigquery_function
+
+
+def test_deploy_udf_with_name():
+    session = mocks.create_bigquery_session()
+
+    def my_remote_func(x: int) -> int:
+        return x * 2
+
+    deployed = session.deploy_udf(my_remote_func, name="my_custom_name")
+
+    # Test that the function would have been deployed somewhere.
+    assert "my_custom_name" in deployed.bigframes_bigquery_function