fix: fix read_gbq_function issue in dataframe apply method

bigframes/dataframe.py

@@ -3959,7 +3959,16 @@ def apply(self, func, *, axis=0, args: typing.Tuple = (), **kwargs):
             return result_series
 
         # Per-column apply
-        results = {name: func(col, *args, **kwargs) for name, col in self.items()}
+        if hasattr(func, "bigframes_remote_function"):
+            if args or kwargs:
+                warnings.warn(
+                    "The args and kwargs are not supported in the remote function.",
+                    category=bigframes.exceptions.ArgsAndKwargsNotSupportedWarning,
+                )
+            results = {name: col.apply(func) for name, col in self.items()}
+        else:
+            results = {name: func(col, *args, **kwargs) for name, col in self.items()}
+
         if all(
             [
                 isinstance(val, bigframes.series.Series) or utils.is_list_like(val)

bigframes/exceptions.py

@@ -77,3 +77,7 @@ class ApiDeprecationWarning(FutureWarning):
 
 class BadIndexerKeyWarning(Warning):
     """The indexer key is not used correctly."""
+
+
+class ArgsAndKwargsNotSupportedWarning(Warning):
+    """The args and kwargs are not supported."""

tests/system/small/test_remote_function.py

@@ -839,6 +839,30 @@ def test_read_gbq_function_enforces_explicit_types(
         )
 
 
+@pytest.mark.flaky(retries=2, delay=120)
+def test_df_apply(session, scalars_dfs):
+    scalars_df, scalars_pandas_df = scalars_dfs
+    bdf = bigframes.pandas.DataFrame(
+        {
+            "Column1": scalars_df["string_col"],
+            "Column2": scalars_df["string_col"],
+        }
+    )
+
+    func_ref = session.read_gbq_function("bqutil.fn.cw_lower_case_ascii_only")
+    bf_result = bdf.apply(func_ref).to_pandas()
+    # The str.lower() method has the same functionality as func_ref.
+    pd_result = pd.DataFrame(
+        {
+            "Column1": scalars_pandas_df["string_col"].str.lower(),
+            "Column2": scalars_pandas_df["string_col"].str.lower(),
+        }
+    )
+
+    pd.testing.assert_series_equal(pd_result["Column1"], bf_result["Column1"])
+    pd.testing.assert_series_equal(pd_result["Column2"], bf_result["Column2"])
+
+
 @pytest.mark.flaky(retries=2, delay=120)
 def test_read_gbq_function_multiple_inputs_not_a_row_processor(session):
     with pytest.raises(ValueError) as context: