fix: fix read_gbq_function issue in dataframe apply method

bigframes/dataframe.py

@@ -3922,6 +3922,10 @@ def map(self, func, na_action: Optional[str] = None) -> DataFrame:
         )
 
     def apply(self, func, *, axis=0, args: typing.Tuple = (), **kwargs):
+        # In Bigframes remote function, DataFrame '.apply' method is specifically
+        # designed to work with row-wise or column-wise operations, where the input
+        # to the applied function should be a Series, not a scalar.
+
         if utils.get_axis_number(axis) == 1:
             msg = "axis=1 scenario is in preview."
             warnings.warn(msg, category=bfe.PreviewWarning)
@@ -4016,8 +4020,26 @@ def apply(self, func, *, axis=0, args: typing.Tuple = (), **kwargs):
             result_series.name = None
             return result_series
 
+        # This is when the func as a remote function is applied to each element of
+        # the dataframe (not supported).
+        if hasattr(func, "is_row_processor") and not func.is_row_processor:
+            raise NotImplementedError(
+                "In Bigframes remote function, DataFrame '.apply()' does not "
+                "support element-wise application. Please use '.map()' instead."
+            )
+
+        # At this point column-wise operation will be performed (not supported).
+        if hasattr(func, "bigframes_remote_function"):
+            raise NotImplementedError(
+                "DataFrame '.apply()' does not support remote function for "
+                "column-wise application (i.e. with axis=0). Please use '.map()' "
+                "instead for element-wise application of the remote function, or "
+                "use regular python function for column-wise application."
+            )
+
         # Per-column apply
         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)

tests/system/small/test_remote_function.py

@@ -839,6 +839,28 @@ def test_read_gbq_function_enforces_explicit_types(
         )
 
 
+@pytest.mark.flaky(retries=2, delay=120)
+def test_df_apply_scalar_func(session, scalars_dfs):
+    scalars_df, _ = scalars_dfs
+    bdf = bigframes.pandas.DataFrame(
+        {
+            "Column1": scalars_df["string_col"],
+            "Column2": scalars_df["string_col"],
+        }
+    )
+
+    # The "cw_lower_case_ascii_only" is a scalar function.
+    func_ref = session.read_gbq_function("bqutil.fn.cw_lower_case_ascii_only")
+
+    # DataFrame '.apply()' only supports series level application.
+    with pytest.raises(NotImplementedError) as context:
+        bdf.apply(func_ref)
+    assert str(context.value) == (
+        "In Bigframes remote function, DataFrame '.apply()' does not support "
+        "element-wise application. Please use '.map()' instead."
+    )
+
+
 @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: