feat: pivot_table supports fill_value arg (#2257)

Thank you for opening a Pull Request! Before submitting your PR, there
are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a
[bug/issue](https://github.com/googleapis/python-bigquery-dataframes/issues/new/choose)
before writing your code! That way we can discuss the change, evaluate
designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

Author: TrevorBergeron

bigframes/core/reshape/pivot.py

@@ -71,12 +71,11 @@ def crosstab(
         columns=tmp_col_names,
         aggfunc=aggfunc or "count",
         sort=False,
+        fill_value=0 if (aggfunc is None) else None,
     )
+    # Undo temporary unique level labels
     pivot_table.index.names = rownames or [i.name for i in index]
     pivot_table.columns.names = colnames or [c.name for c in columns]
-    if aggfunc is None:
-        # TODO: Push this into pivot_table itself
-        pivot_table = pivot_table.fillna(0)
     return pivot_table
 
 

bigframes/dataframe.py

@@ -3486,10 +3486,6 @@ def pivot_table(
         observed: bool = False,
         sort: bool = True,
     ) -> DataFrame:
-        if fill_value is not None:
-            raise NotImplementedError(
-                "DataFrame.pivot_table fill_value arg not supported. {constants.FEEDBACK_LINK}"
-            )
         if margins:
             raise NotImplementedError(
                 "DataFrame.pivot_table margins arg not supported. {constants.FEEDBACK_LINK}"
@@ -3549,14 +3545,16 @@ def pivot_table(
             index=index,
             values=values if len(values) > 1 else None,
         )
+        if fill_value is not None:
+            pivoted = pivoted.fillna(fill_value)
         if sort:
             pivoted = pivoted.sort_index()
 
         # TODO: Remove the reordering step once the issue is resolved.
         # The pivot_table method results in multi-index columns that are always ordered.
         # However, the order of the pivoted result columns is not guaranteed to be sorted.
         # Sort and reorder.
-        return pivoted[pivoted.columns.sort_values()]
+        return pivoted.sort_index(axis=1)  # type: ignore
 
     def stack(self, level: LevelsType = -1):
         if not isinstance(self.columns, pandas.MultiIndex):

tests/system/small/test_dataframe.py

@@ -3784,12 +3784,18 @@ def test_df_pivot_hockey(hockey_df, hockey_pandas_df, values, index, columns):
 
 
 @pytest.mark.parametrize(
-    ("values", "index", "columns", "aggfunc"),
+    ("values", "index", "columns", "aggfunc", "fill_value"),
     [
-        (("culmen_length_mm", "body_mass_g"), "species", "sex", "std"),
-        (["body_mass_g", "culmen_length_mm"], ("species", "island"), "sex", "sum"),
-        ("body_mass_g", "sex", ["island", "species"], "mean"),
-        ("culmen_depth_mm", "island", "species", "max"),
+        (("culmen_length_mm", "body_mass_g"), "species", "sex", "std", 1.0),
+        (
+            ["body_mass_g", "culmen_length_mm"],
+            ("species", "island"),
+            "sex",
+            "sum",
+            None,
+        ),
+        ("body_mass_g", "sex", ["island", "species"], "mean", None),
+        ("culmen_depth_mm", "island", "species", "max", -1),
     ],
 )
 def test_df_pivot_table(
@@ -3799,12 +3805,21 @@ def test_df_pivot_table(
     index,
     columns,
     aggfunc,
+    fill_value,
 ):
     bf_result = penguins_df_default_index.pivot_table(
-        values=values, index=index, columns=columns, aggfunc=aggfunc
+        values=values,
+        index=index,
+        columns=columns,
+        aggfunc=aggfunc,
+        fill_value=fill_value,
     ).to_pandas()
     pd_result = penguins_pandas_df_default_index.pivot_table(
-        values=values, index=index, columns=columns, aggfunc=aggfunc
+        values=values,
+        index=index,
+        columns=columns,
+        aggfunc=aggfunc,
+        fill_value=fill_value,
     )
     pd.testing.assert_frame_equal(
         bf_result, pd_result, check_dtype=False, check_column_type=False

third_party/bigframes_vendored/pandas/core/frame.py

@@ -6414,6 +6414,10 @@ def pivot_table(self, values=None, index=None, columns=None, aggfunc="mean"):
             aggfunc (str, default "mean"):
                 Aggregation function name to compute summary statistics (e.g., 'sum', 'mean').
 
+            fill_value (scalar, default None):
+                Value to replace missing values with (in the resulting pivot table, after
+                aggregation).
+
         Returns:
             bigframes.pandas.DataFrame: An Excel style pivot table.
         """